Source code for sofar.sofa

import os
import re
import json
from datetime import datetime
import platform
import numpy as np
import warnings
from packaging.version import parse
from copy import deepcopy
import sofar as sf
from .utils import (_nd_newaxis, _atleast_nd, _get_conventions,
                    _verify_convention_and_version)


[docs] class Sofa(): """Create a new SOFA object. Parameters ---------- convention : str The name of the convention from which the SOFA file is created. See :py:func:`~sofar.list_conventions`. mandatory : bool, optional If ``True``, only the mandatory data of the convention will be returned. The default is ``False``, which returns mandatory and optional data. version : str, optional The version of the convention as a string, e.g., ``' 2.0'``. The default is ``'latest'``. Also see :py:func:`~sofar.list_conventions`. verify : bool, optional Verify the SOFA object by calling :py:func:`~Sofa.verify`. This helps to find potential errors in the default values and is thus recommended If creating a file does not work, try to call `Sofa` with ``verify=False``. The default ``'auto'`` defaults to ``True`` for stable conventions with versions of 1.0 or higher and to ``False`` otherwise. Returns ------- sofa : Sofa A SOFA object filled with the default values of the convention. Examples -------- Create a new SOFA object with default values .. code-block:: python import sofar as sf # create SOFA object sofa = sf.Sofa("SimpleFreeFieldHRIR") Add data as a list .. code-block:: python sofa.Data_IR = [1, 1] Data can be entered as numbers, numpy arrays or lists. Note the following 1. Lists are converted to numpy arrays with at least two dimensions, i.e., ``sofa.Data_IR`` is converted to a numpy array of shape (1, 2) 2. Missing dimensions are appended when writing the SOFA object to disk, i.e., ``sofa.Data_IR`` is written as an array of shape (1, 2, 1) because the SOFA standard AES69 defines it as a three dimensional array with the dimensions (`M: measurements`, `R: receivers`, `N: samples`) 3. When reading data from a SOFA file, array data is always returned as numpy arrays and singleton trailing dimensions are discarded (numpy default). I.e., ``sofa.Data_IR`` will again be an array of shape (1, 2) after writing and reading to and from disk. 4. One dimensional arrays with only one element will be converted to scalar values. E.g. ``sofa.Data_SamplingRate`` is stored as an array of shape (1, ) inside SOFA files (according to the SOFA standard AES69) but will be a scalar inside SOFA objects after reading from disk. For more examples refer to the `Quick tour of SOFA and sofar` at https://sofar.readthedocs.io/en/stable/ """ # these have to be set here, because they are used in __setattr__ and # Python checks if they exist upon class creation # don't allow adding attributes and deleting/writing read only attributes _protected = False # list of read only attributes (filled upon init) _read_only_attr = [] def __init__(self, convention, mandatory=False, version="latest", verify='auto'): """See class docstring""" # get convention if convention is not None: self._convention = self._load_convention(convention, version) # update read only attributes self._read_only_attr = [ key for key in self._convention.keys() if self._read_only(self._convention[key]["flags"])] # add attributes with default values self._convention_to_sofa(mandatory) # set default for verify version = \ self._convention['GLOBAL_SOFAConventionsVersion']['default'] if verify == 'auto': verify = True if parse(version) >= parse('1.0') else False # add and update the API # (mandatory=False can not be verified because some conventions # have default values that have optional variables as dependencies) if verify and not mandatory: self.verify(mode="read") # warning for preliminary conventions if verification is bypassed elif parse(version) < parse('1.0'): warnings.warn(UserWarning(( f"Detected preliminary conventions version {version}. " "Upgrade data to version >= 1.0 if possible. Preliminary " "conventions might change in the future, which could " "invalidate data that was written before the changes."))) self.protected = True else: verify = False self._convention = {} def __setattr__(self, name: str, value): # don't allow new attributes to be added outside the class if self.protected and not hasattr(self, name): raise TypeError(f"{name} is an invalid attribute") # don't allow setting read only attributes if name in self._read_only_attr and self.protected: raise TypeError(( f"{name} is a read only attribute. Iy you know what you are " "doing, you can set Sofa.protected = False to write read " "only data (e.g., to repair corrupted SOFA data).")) # convert to numpy array or scalar if not isinstance(value, (str, dict, np.ndarray)) \ and name != "protected": value = np.atleast_2d(value) if value.size == 1: value = value.flatten()[0] super.__setattr__(self, name, value) def __delattr__(self, name: str): # can't delete non existing attributes if not hasattr(self, name): raise TypeError(f"{name} is not an attribute") # delete anything if not frozen, delete non mandatory if not self.protected or \ not self._mandatory(self._convention[name]["flags"]): super().__delattr__(name) # check if custom field as to be deleted if hasattr(self, "_custom"): if name in self._custom: self._custom.pop(name) else: raise TypeError( f"{name} is a mandatory attribute that can not be deleted") def __repr__(self): return (f"sofar.SOFA object: {self.GLOBAL_SOFAConventions} " f"{self.GLOBAL_SOFAConventionsVersion}") @property def list_dimensions(self): """ Print the dimensions of the SOFA object See :py:func:`~Sofa.inspect` to see the shapes of the data inside the SOFA object and :py:func:`~Sofa.get_dimension` to get the size/value of a specific dimensions as integer number. The SOFA file standard defines the following dimensions that are used to define the shape of the data entries: M number of measurements N number of samples, frequencies, SOS coefficients (depending on self.GLOBAL_DataType) R Number of receivers or SH coefficients (depending on ReceiverPosition_Type) E Number of emitters or SH coefficients (depending on EmitterPosition_Type) S Maximum length of a string in a string array C Size of the coordinate dimension. This is always three. I Single dimension. This is always one. """ # Check if the dimensions can be updated self._update_dimensions() # get verbose description for dimension N if self.GLOBAL_DataType.startswith("FIR"): N_verbose = "samples" elif self.GLOBAL_DataType.startswith("TF"): N_verbose = "frequencies" elif self.GLOBAL_DataType.startswith("SOS"): N_verbose = "SOS coefficients" # get verbose description for dimensions R and E R_verbose = "receiver spherical harmonics coefficients" if \ 'harmonic' in self.ReceiverPosition_Type else "receiver" E_verbose = "emitter spherical harmonics coefficients" if \ 'harmonic' in self.EmitterPosition_Type else "emitter" dimensions = { "M": "measurements", "N": N_verbose, "R": R_verbose, "E": E_verbose, "S": "maximum string length", "C": "coordinate dimensions, fixed", "I": "single dimension, fixed"} info_str = "" for key, value in self._api.items(): dim_info = dimensions[key] if key in dimensions \ else "custom dimension" info_str += f"{key} = {value} {dim_info}" if dim_info != "custom ": for key2, value2 in self._convention.items(): dim = value2["dimensions"] if dim is not None and key.lower() in dim: info_str += \ f" (set by {key2} of dimension {dim.upper()})" break info_str += "\n" print(info_str)
[docs] def get_dimension(self, dimension): """ Get size of a SOFA dimension SOFA dimensions specify the shape of the data contained in a SOFA object. For a list of all dimensions see :py:func:`~list_dimensions`. Parameters ---------- dimension : str The dimension as a string, e.g., ``'N'``. Returns ------- size : int the size of the queried dimension. """ # Check if the dimensions can be updated self._update_dimensions() if dimension not in self._api: raise ValueError(( f"{dimension} is not a valid dimension. " "See Sofa.list_dimensions for a list of valid dimensions.")) return self._api[dimension]
def _update_dimensions(self): """ Call verify and raise an error if the dimensions could not be updated. used in Sofa.list_dimensions and Sofa.get_dimension """ issues = self.verify(issue_handling="return") if issues is not None and ("data of wrong type" in issues or "variables of wrong shape" in issues or not hasattr(self, "_api")): raise ValueError(("Dimensions can not be shown because variables " "of wrong type or shape were detected. " "Call Sofa.verify() for more information."))
[docs] def info(self, info="all"): """ Print information about the convention of a SOFA object. Prints the variable type (attribute, double, string), shape, flags (mandatory, read only) and comment (if any) for each or selected entries. Parameters ---------- info : str Specifies the kind of information that is printed: ``'all'`` ``'mandatory'`` ``'optional'`` ``'read only'`` ``'data'`` Print the name, type, shape, and flags and comment for all or selected entries of the SOFA object. ``'data'`` does not show entries of type attribute. key If key is the name of an object attribute, all information for attribute will be printed. """ # warn for upcoming deprecation warnings.warn(( 'Sofa.info() will be deprecated in sofar 1.3.0 The conventions are' ' now documented at ' 'https://sofar.readthedocs.io/en/stable/resources/conventions.html'), # noqa UserWarning) # update the private attribute `_convention` to make sure the required # meta data is in place if not hasattr(self, "_convention"): self._reset_convention() # list of all attributes keys = [k for k in self.__dict__.keys() if not k.startswith("_")] # start printing the information info_str = ( f"{self.GLOBAL_SOFAConventions} " F"{self.GLOBAL_SOFAConventionsVersion} " f"(SOFA version {self.GLOBAL_Version})\n") info_str += "-" * len(info_str) + "\n" if info in ["all", "mandatory", "optional", "read only", "data"]: info_str += f"showing {info} entries : type (shape), flags\n\n" for key in keys: # check if field should be skipped flags = self._convention[key]["flags"] if (not self._mandatory(flags) and info == "mandatory") \ or \ (self._mandatory(flags) and info == "optional") \ or \ (not self._read_only(flags) and info == "read only") \ or \ (self._convention[key]['type'] == "attribute" and info == "data"): continue info_str += f"{key} : {self._convention[key]['type']}" if self._convention[key]['dimensions']: info_str += \ f" ({self._convention[key]['dimensions'].upper()})" if self._mandatory(flags): info_str += ", mandatory" else: info_str += ", optional" if self._read_only(flags): info_str += ", read only" if self._convention[key]['comment']: info_str += f"\n {self._convention[key]['comment']}\n" else: info_str += "\n" elif info in keys: for key in [k for k in keys if info in k]: comment = str(self._convention[key]['comment']) info_str += ( f"{key}\n" f" type: {self._convention[key]['type']}\n" f" mandatory: " f"{self._mandatory(self._convention[key]['flags'])}\n" f" read only: " f"{self._read_only(self._convention[key]['flags'])}\n" f" default: {self._convention[key]['default']}\n" f" shape: " f"{str(self._convention[key]['dimensions']).upper()}\n" f" comment: {comment}\n") else: raise ValueError(f"info='{info}' is invalid") print(info_str)
[docs] def inspect(self, file=None, issue_handling="print"): """ Get information about data inside a SOFA object. Prints the values of all attributes and variables with six or less entries and the shapes and type of all numeric and string variables. When printing the values of arrays, single dimensions are discarded for easy of display, i.e., an array of shape (1, 3, 2) will be displayed as an array of shape (3, 2). Parameters ---------- file : str Full path of a file under which the information is to be stored in plain text. The default ``None`` only print the information to the console. issue_handling : str, optional Defines how issues detected during verification of the SOFA object are handled (see :py:func:`~sofar.sofar.Sofa.verify`) ``'raise'`` Warnings and errors are raised if issues are detected ``'print'`` Issues are printed without raising warnings and errors ``'return'`` Issues are returned as string but neither raised nor printed ``'ignore'`` Issues are ignored, i.e., not raised, printed, or returned. The default is ``'print'``. """ # update the private attribute `_convention` to make sure the required # meta data is in place self.verify(issue_handling=issue_handling) # list of all attributes keys = [k for k in self.__dict__.keys() if not k.startswith("_")] # start printing the information info_str = ( f"{self.GLOBAL_SOFAConventions} " F"{self.GLOBAL_SOFAConventionsVersion} " f"(SOFA version {self.GLOBAL_Version})\n") info_str += "-" * len(info_str) + "\n" for key in keys: info_str += key + " : " value = getattr(self, key) # information for attributes and scalars if self._convention[key]["type"] == "attribute" or value.size == 1: info_str += str(value) + "\n" # information for variables else: # get shape and dimension shape = value.shape dimension = self._dimensions[key] # pad shape if required (trailing single dimensions are # discarded following the numpy default) while len(shape) < len(dimension): shape += (1, ) # make verbose shape, e.g., '(M=100, R=2, N=128, ' shape_verbose = "(" for s, d in zip(shape, dimension): shape_verbose += f"{d}={s}, " # add shape information info_str += shape_verbose[:-2] + ")\n" # add value information if not too much if value.size < 7: info_str += " " + \ str(np.squeeze(value)).replace("\n", "\n ") + "\n" # write to text file if file is not None: with open(file, 'w') as f_id: f_id.write(info_str + "\n") # output to console print(info_str)
[docs] def add_missing(self, mandatory=True, optional=True, verbose=True): """ Add missing data with default values. Data might be missing in SOFA objects if the creator did not include it or if a new data was suggested for a newer version of a SOFA convention. Use this function to add the data with its default values. mandatory : Bool Add missing mandatory data. The default is ``True``. optional : Bool Add missing optional data. The default is ``True``. verbose : Bool Print the information about added data to the console. The default is ``True``. """ # initialize self._reset_convention() added = "Added the following missing data with their default values:\n" # current data keys = [key for key in self.__dict__.keys() if not key.startswith("_")] self.protected = False # loop data in convention for key in self._convention.keys(): is_mandatory = self._mandatory(self._convention[key]["flags"]) add_data = (is_mandatory and mandatory) or \ (not is_mandatory and optional) # add with default if key not in keys and add_data: setattr(self, key, self._convention[key]["default"]) added += f"- {key} " added += f"({'mandatory' if is_mandatory else 'optional'})\n" self.protected = True if verbose: if "-" in added: print(added) else: print("All mandatory data contained.")
[docs] def add_variable(self, name, value, dtype, dimensions): """ Add custom variable to the SOFA object, i.e., numeric or string arrays. Parameters ---------- name : str Name of the new variable. value : any value to be added (see `dtype` for restrictions). dtype : str Type of the entry to be added in netCDF style: ``'double'`` Use this to store numeric data that can be provided as number list or numpy array. ``'string'`` Use this to store string variables as numpy string arrays of type ``'U'`` or ``'S'``. dimensions : str The shape of the new entry as a string. See :py:func:`~Sofa.list_dimensions`. Examples -------- .. code-block:: python import sofar as sf sofa = sf.Sofa("GeneralTF") # add numeric data sofa.add_variable("Temperature", 25.1, "double", "MI") # add GLOBAL and Variable attribute sofa.add_entry( "GLOBAL_DateMeasured", "8.08.2021", "attribute", None) sofa.add_entry( "Temperature_Units", "degree Celsius", "attribute", None) # add a string data sofa.add_variable( "Comment", "Measured with wind screen", "string", "MS") """ self._add_entry(name, value, dtype, dimensions)
[docs] def add_attribute(self, name, value): """ Add custom attribute to the SOFA object. Parameters ---------- name : str Name of the new attribute. value : str value to be added. Examples -------- .. code-block:: python import sofar as sf sofa = sf.Sofa("GeneralTF") # add GLOBAL and Variable attribute sofa.add_attribute("GLOBAL_DateMeasured", "8.08.2021") sofa.add_attribute("Data_Real_Units", "Pascal") """ self._add_entry(name, value, 'attribute', None)
[docs] def delete(self, name): """ Delete variable or attribute from SOFA object. Note that mandatory data can not be deleted. Check the `sofar documentation <https://sofar.readthedocs.io/en/stable/resources/conventions.html>`_ for a complete list of optional variables and attributes. Parameters ---------- name : str Name of the variable or attribute to be deleted """ delattr(self, name)
@property def protected(self): """ If Sofa.protected is ``True``, read only data can not be changed. Only change this to ``False`` if you know what you are doing, e.g., if you need to repair corrupted SOFA data. """ return self._protected @protected.setter def protected(self, value: bool): """ If Sofa.protected is ``True``, read only data can not be changed. Only change this to ``False`` if you know what you are doing, e.g., if you need to repair corrupted SOFA data. """ if not isinstance(value, bool): raise ValueError("Sofa.protected can only be True or False") self._protected = value def _add_entry(self, name, value, dtype, dimensions): """ Add custom data to a SOFA object. See add_variable and add_attribute for more information. """ # check input if hasattr(self, name): raise ValueError(f"Entry {name} already exists") if dtype not in ["attribute", "double", "string"]: raise ValueError( f"dtype is {dtype} but must be attribute, double, or string") if "_" in name and dtype != "attribute": raise ValueError(("underscores '_' in the name are only " "allowed for attributes")) if dtype == "attribute": if name.count("_") != 1 or \ (name.startswith("Data_") and (name.count("_") == 0 or name.count("_") > 2)): raise ValueError((f"The name of {name} must have the " "form VariableName_AttributeName")) if not name.startswith("GLOBAL_") and \ name[:name.rindex("_")] not in self._convention: raise ValueError((f"Adding Attribute {name} requires " f"variable {name[:name.rindex('_')]}")) if dimensions is None and dtype != "attribute": raise ValueError(("dimensions must be provided for entries of " "dtype double and string")) if dimensions is not None: dimensions = dimensions.upper() for dimension in dimensions: if dimension not in "ERMNCIS": warnings.warn( f"Added custom dimension {dimensions} to SOFA object") # add attribute to class self._add_custom_api_entry(name, value, None, dimensions, dtype) def _add_custom_api_entry(self, key, value, flags, dimensions, dtype): """ Add custom entry to the sofa._convention and permanently save it in sofa._custom Parameters ---------- key : str name of the entry value: any Value of the entry flags, dimensions, dtype : any as in sofa._convention dimensions : string Dimensions in case of numeric or string array dtype : string double, string, or attribute """ # create custom API if it not exists self.protected = False # lower case letters to indicate custom dimensions if dimensions is not None: dimensions = [d.upper() if d.upper() in "ERMNCIS" else d.lower() for d in dimensions] dimensions = "".join(dimensions) # add user entry to custom API, if not contained in the convention if not hasattr(self, "_convention") or key not in self._convention: if not hasattr(self, "_custom"): self._custom = {} self._custom[key] = { "flags": flags, "dimensions": dimensions, "type": dtype, "default": None, "comment": ""} self._convention[key] = self._custom[key] # add attribute to object setattr(self, key, value) self.protected = True
[docs] def upgrade_convention(self, target=None, verify='auto'): """ Upgrade Sofa data to newer conventions. Calling this with the default arguments returns a list of possible conventions to which the data will be upgraded. If the data is up to date the list will be empty. Parameters ---------- target : str, optional The convention and version to which the data should be upgraded as a string. For example ``'SimpleFreeFieldHRIR_1.0'`` would upgrade the data to the SOFA-Convention `SimpleFreeFieldHRIR` version 1.0. The default is ``None`` which returns a list of possible conventions to which the data can be updated. verify : bool, optional Flag to specify if the data should be verified after the upgrade using :py:func:`~Sofa.verify`. The default ``'auto'`` defaults to ``True`` for stable conventions with versions of 1.0 or higher and to ``False`` otherwise. Returns ------- target : list of strings List with available conventions to which the data can be updated. If the data is up to data, the list will be empty. `target` is only returned if `target` is ``None``. """ # check input --------------------------------------------------------- self._reset_convention() # get deprecations and information about Sofa object _, _, deprecations, upgrade = self._verification_rules() convention_current = self.GLOBAL_SOFAConventions version_current = self.GLOBAL_SOFAConventionsVersion sofa_version_current = self.GLOBAL_Version # check if convention is deprecated ----------------------------------- is_deprecated = False if convention_current in deprecations["GLOBAL:SOFAConventions"]: is_deprecated = True elif convention_current in upgrade: for from_to in upgrade[convention_current]["from_to"]: if version_current in from_to[0]: is_deprecated = True break # check for upgrades -------------------------------------------------- if is_deprecated: # check if upgrade is available for this convention if convention_current not in upgrade: print((f"Convention {convention_current} v{version_current} is" " outdated but is missing upgrade rules")) return # check if upgrade is available for this version for from_to in upgrade[convention_current]["from_to"]: if version_current in from_to[0]: targets = from_to[1] if target in targets: target_id = from_to[2] else: if target is not None: print(f"{target} is invalid.") upgrades = ( f"{convention_current} v{version_current} " "can be upgraded to:\n") for t in targets: t = t.split("_") upgrades += f"- {t[0]} v{t[1]}\n" print(upgrades) return targets break else: print((f"Convention {convention_current} v{version_current} " "is up to date")) return # get information to upgrade ------------------------------------------ upgrade = upgrade[self.GLOBAL_SOFAConventions][target_id] convention_target, version_target = target.split("_") # upgrade ------------------------------------------------------------- self._convention = self._load_convention( convention_target, version_target) sofa_version_target = self._convention["GLOBAL_Version"]["default"] print((f"Upgrading {convention_current} v{version_current} to " f"{convention_target} v{version_target} (SOFA version " f"{sofa_version_current} to {sofa_version_target})")) # upgrade convention keys = ["GLOBAL_SOFAConventions", "GLOBAL_SOFAConventionsVersion", "GLOBAL_Version", "GLOBAL_DataType"] self.protected = False for key in keys: setattr(self, key, self._convention[key]["default"]) # move data for source, move in upgrade["move"].items(): # get info source_sofar = source.replace(".", '_').replace(":", "_") target_sofar = move["target"].replace(".", '_').replace(":", "_") move_info = f"- Moving {source_sofar} to {target_sofar}." # get data data = getattr(self, source_sofar) # delete from Sofa object delattr(self, source_sofar) # Check move axis moveaxis = upgrade["move"][source]["moveaxis"] if moveaxis is not None: # add dimensions if required # (sofar discards trailing singular dimensions) while data.ndim < np.max(moveaxis) + 1: data = data[..., None] # move the axis data = np.moveaxis(data, moveaxis[0], moveaxis[1]) move_info += f" Moving axis {moveaxis[0]} to {moveaxis[1]}." # Check deprecated dimensions deprecated_dimensions = \ upgrade["move"][source]["deprecated_dimensions"] if deprecated_dimensions is not None: move_info += ( f" WARNING: Dimensions {', '.join(deprecated_dimensions)} " "are now deprecated.") # add data setattr(self, target_sofar, data) # print info print(move_info) if not upgrade["move"]: print("- No data to move") # remove data for target in upgrade["remove"]: target_sofar = target.replace(".", '_').replace(":", "_") delattr(self, target_sofar) print(f"- Deleting {target_sofar}.") if not upgrade["remove"]: print("- No data to remove") # check for missing mandatory data self.add_missing(True, False) self.protected = True # display general message if upgrade["message"] is not None: print(upgrade["message"]) # set default for verify if verify == 'auto': version = self.GLOBAL_SOFAConventionsVersion verify = True if parse(version) >= parse('1.0') else False if verify: self.verify()
[docs] def verify(self, issue_handling="raise", mode="write"): """ Verify a SOFA object against the SOFA standard. This function updates the API, and checks the following - Are all mandatory data contained? If `issue_handling` is ``"raise"`` missing mandatory data raises an error. Otherwise mandatory data are added with their default value and a warning is given. - Are the names of variables and attributes in accordance to the SOFA standard? - Are the data types in accordance with the SOFA standard? - Are the dimensions of the variables consistent and in accordance to the SOFA standard? - Are the values of attributes consistent and in accordance to the SOFA standard? A detailed set of validation rules can be found at https://github.com/pyfar/sofar/tree/main/sofar/verification_rules .. note:: :py:func:`~verify` is automatically called when you create a new SOFA object, read a SOFA file from disk, and write a SOFA file to disk (using the default parameters). The API of a SOFA object consists of four parts, that are stored dictionaries in private attributes. This is required for writing data with :py:func:`~sofa.write_sofa` and should usually not be manipulated outside of :py:func:`~verify` self._convention The SOFA convention with default values, variable dimensions, flags and comments. These data are read from the official SOFA conventions contained in the SOFA Matlab/Octave API. self._dimensions The detected dimensions of the data inside the SOFA object. self._api The size of the dimensions (see py:func:`~list_dimensions`). This specifies the dimensions of the data inside the SOFA object. self._custom Stores information of custom variables that are not defined by the convention. The format is the same as in `self._convention`. Parameters ---------- issue_handling : str, optional Defines how detected issues are handled ``'raise'`` Warnings and errors are raised if issues are detected ``'print'`` Issues are printed without raising warnings and errors ``'return'`` Issues are returned as string but neither raised nor printed The default is ``'raise'``. mode : str, optional The SOFA standard is more strict for writing data than for reading data. ``'write'`` All units (e.g. 'meter') must be written be lower case. ``'read'`` Units can contain upper case letters (e.g. 'Meter') The default is ``'write'`` Returns ------- issues : str, None Detected issues as a string. None if no issues were detected. Note that this is only returned if ``issue_handling='return'`` (see above) """ # NOTE: This function collects warnings and errors and tries to output # them in a block. This makes the code slightly more complicated but # is more convenient for the user and with respect to a potential # future web based tool for verifying SOFA files. # initialize warning and error messages error_msg = "\nERRORS\n------\n" warning_msg = "\nWARNINGS\n--------\n" # --------------------------------------------------------------------- # 0. update the convention self._reset_convention() # --------------------------------------------------------------------- # 1. check if the mandatory attributes are contained missing = "" keys = [key for key in self.__dict__.keys() if not key.startswith("_")] for key in self._convention.keys(): if self._mandatory(self._convention[key]["flags"]) \ and key not in keys: if issue_handling != "raise": # add missing data with default value self.protected = False setattr(self, key, self._convention[key]["default"]) self.protected = True # prepare to raise warning missing += "- " + key + "\n" if missing: if issue_handling == "raise": error_msg += ("Detected missing mandatory data " "call sofa.add_missing() to fix this):\n") error_msg += missing else: warning_msg += "Added mandatory data with default values:\n" warning_msg += missing # --------------------------------------------------------------------- # 2. verify data type current_error = "" for key in keys: # handle dimensions dimensions = self._convention[key]["dimensions"] dtype = self._convention[key]["type"] # check data type value = getattr(self, key) if dtype == "attribute": if not isinstance(value, str): current_error += \ f"- {key} must be string but is {type(value)}\n" elif dtype == "double": # All variables can be of type float or int, and are converted # to float when writing. # sofar does not force the user to pass data as numpy arrays. # We thus check for allowed instances (int, float, numpy) and # specifically for the type of numpy arrays. if not isinstance( value, (int, float, np.ndarray, np.number)) or \ (isinstance(value, (np.ndarray, np.number)) and value.dtype.kind not in ['i', 'f']): current_error += (f"- {key} must be int or float " f"but is {type(value)}\n") elif dtype == "string": # multiple checks needed because sofar does not force the user # to initially pass data as numpy arrays if not isinstance(value, (str, np.ndarray)): current_error += (f"- {key} must be string or numpy array " f"but is {type(value)}\n") if isinstance(value, np.ndarray) and not ( str(value.dtype).startswith('<U') or str(value.dtype).startswith('<S')): current_error += (f"- {key} must be U or S " f"but is {type(value.dtype)}\n") else: # Could only be tested by manipulating JSON convention files # (Could take different data types in the future and convert to # numpy double arrays.) current_error += ( f"- {key}: Error in convention. Type must be " f"double, string, or attribute but is {dtype}\n") if current_error: error_msg += "Detected data of wrong type:\n" error_msg += current_error # if an error ocurred up to here, it has to be handled. Otherwise # detecting the dimensions might fail. Warnings are not reported until # the end if error_msg != "\nERRORS\n------\n" and issue_handling != "ignore": _, issues = self._verify_handle_issues( "\nWARNINGS\n--------\n", error_msg, issue_handling) if issue_handling == "print": return else: # (issue_handling == "return"): return issues # --------------------------------------------------------------------- # 3. Verify names of entries # check attributes without variables current_error = "" for key in keys: if self._convention[key]["type"] != "attribute" or \ key.count("_") == 0: continue if (key[:key.rindex("_")] not in self._convention and not key.startswith("GLOBAL_")): current_error += "- " + key + "\n" if current_error: error_msg += "Detected attributes with missing variables:\n" error_msg += current_error # check number of underscores current_error = "" for key in keys: if self._convention[key]["type"] != "attribute": continue # the case above caught attributes with too many underscores if key.count("_") == 0: current_error += "- " + key + "\n" if current_error: error_msg += ( "Detected attribute names with too many or little underscores." " Names must have the form Variable_Attribute, Data_Attribute " "(one underscore), or Data_Variable_Attribute (two " "underscores):\n") error_msg += current_error # check numeric variables current_error = "" for key in keys: if self._convention[key]["type"] == "attribute": continue if "_" in key.replace("Data_", ""): current_error += "- " + key + "\n" if current_error: error_msg += ( "Detected variable names with too many underscores." "Underscores are only allowed for the variable Data:\n") error_msg += current_error # check reserved names current_error = "" for key in keys: # AES69 Sec. 4.7.1 if key.startswith("PRIVATE") or key.startswith("API"): current_error += "- " + key + "\n" if (key.startswith("GLOBAL") and not key.startswith("GLOBAL_")) or\ (key.startswith("GLOBAL") and self._convention[key]["type"] != "attribute"): current_error += "- " + key + "\n" if current_error: error_msg += ( "Detected variable or attribute with reserved key words " "PRIVATE, API, or GLOBAL:\n") error_msg += current_error # check names of custom data (shall not have the same name as # normative data contained in the convention AES69-2022 Sec. 5.3) current_error = "" if hasattr(self, "_custom"): # get normative variables and attributes normative = sf.Sofa( self.GLOBAL_SOFAConventions, version=self.GLOBAL_SOFAConventionsVersion)._convention.keys() normative = [n.replace("GLOBAL_", "") for n in normative] # compare against custom for key in self._custom.keys(): if key.replace("GLOBAL_", "") in normative: current_error += "- " + key + "\n" if current_error: error_msg += ( "Detected custom variable or attribute with reserved names. " "Custom data shall not have the same name as data contained in" " the convention itself:\n") error_msg += current_error # --------------------------------------------------------------------- # 4. Get dimensions (E, R, M, N, S, c, I, and custom) # initialize required API fields self.protected = False self._dimensions = {} self._api = {} self.protected = True # get keys for checking the dimensions (all SOFA variables) keys = [key for key in self.__dict__.keys() if key in self._convention and self._convention[key]["dimensions"] is not None] if hasattr(self, "_custom"): keys_custom = [key for key in self._custom.keys() if not key.startswith("_") and self._custom[key]["dimensions"] is not None] keys += keys_custom S = 0 for key in keys: value = getattr(self, key) dimensions = self._convention[key]["dimensions"] # - dimensions are given as string, e.g., 'mRN', or 'IC, MC' # - defined by lower case letters in `dimensions` for id, dim in enumerate(dimensions.split(", ")[0]): if dim not in "ICS" and dim.islower(): # numeric data self._api[dim.upper()] = \ _nd_newaxis(value, 4).shape[id] if dim == "S": # string data S = max(S, np.max(self._get_size_and_shape_of_string_var( value, key)[0])) # add fixed sizes self._api["C"] = 3 self._api["I"] = 1 self._api["S"] = S # --------------------------------------------------------------------- # 5. verify dimensions of data current_error = "" for key in keys: # handle dimensions dimensions = self._convention[key]["dimensions"] dtype = self._convention[key]["type"] # get value and actual shape try: value = getattr(self, key).copy() except AttributeError: value = getattr(self, key) if dtype in ["attribute", "string"]: # string or string array like data shape_act = self._get_size_and_shape_of_string_var( value, key)[1] elif len(dimensions.split(",")[0]) > 1: # multidimensional array like data shape_act = _atleast_nd(value, 4).shape else: # scalar of single dimensional array like data shape_act = (np.array(value).size, ) shape_matched = False for dim in dimensions.split(", "): # get the reference shape ('S' translates to a shape of 1, # because the strings are stored in an array whose shape does # not reflect the max. lengths of the actual strings inside it) shape_ref = tuple( [self._api[d.upper()] if d != "S" else 1 for d in dim]) # get shape for comparison to correct length by cropping and # appending singleton dimensions if required shape_compare = shape_act[:len(shape_ref)] for _ in range(len(shape_ref) - len(shape_compare)): shape_compare += (1, ) # check if the shapes match and write to API if shape_compare == shape_ref: shape_matched = True self._dimensions[key] = dim.upper() break if not shape_matched: # get possible dimensions in verbose form, i.e., "(M=2, C=3)"" dimensions_verbose = [] for dim in dimensions.upper().replace(" ", "").split(","): dimensions_verbose.append( f"({', '.join([f'{d}={self._api[d]}' for d in dim])})") current_error += ( f"- {key} has shape {shape_compare} but must " f"have {', '.join(dimensions_verbose)}\n") if current_error: error_msg += "Detected variables of wrong shape:\n" error_msg += current_error # --------------------------------------------------------------------- # 6. check restrictions on the content of SOFA files rules, unit_aliases, deprecations, _ = self._verification_rules() current_error = "" for key in rules.keys(): # convert to sofar format key_sofar = key.replace(".", "_").replace(":", "_") if not hasattr(self, key_sofar): continue # actual and possible values for the current key test = getattr(self, key_sofar) ref = rules[key]["value"] # test if the value is valid if not self._verify_value(test, ref, unit_aliases, key_sofar): current_error += (f"- {key_sofar} is {test} " f"but must be {', '.join(ref)}\n") # get lower case value for of test for verifying specific # dependencies if isinstance(test, str) and test.lower() in \ ["cartesian", "spherical", "spherical harmonics"]: test = test.lower() # check general dependencies items = rules[key]["general"] if "general" in rules[key] else [] for key_dep in items: # convert key to sofar format key_dep = key_dep.replace(".", "_").replace(":", "_") # check if dependency is contained in SOFA object hard to test, # for mandatory fields (added by sofar by default). if not hasattr(self, key_dep): current_error += (f"- {key_dep} must be given if " f"{key_sofar} is in SOFA object\n") continue # check specific dependencies if "specific" in rules[key] \ and test in rules[key]["specific"]: items = rules[key]["specific"][test].items() else: items = {}.items() for key_dep, ref_dep in items: if key_dep == "_dimensions": # requires specific dimension(s) to have a certain size for dim in rules[key]["specific"][test]["_dimensions"]: # possible sizes dim_ref = \ rules[key]["specific"][test]["_dimensions"][dim]["value"] # noqa # current size dim_act = self._api[dim] # verbose error string for possible sizes dim_str = \ rules[key]["specific"][test]["_dimensions"][dim]["value_str"] # noqa # perform the check if dim_act not in dim_ref: current_error += \ (f"- Dimension {dim} is of size {dim_act} " f"but must be {dim_str} if {key_sofar} " f"is {test}\n") else: # convert name from NetCDF format to sofar format key_dep_sofar = key_dep.replace(".", "_").replace(":", "_") # check if dependency is contained in SOFA object if not hasattr(self, key_dep_sofar): current_error += (f"- {key_dep_sofar} must be given if" f" {key_sofar} is {test}\n") continue # check if dependency has the correct value if ref_dep is None: continue # convert name from NetCDF format to sofar format test_dep = getattr(self, key_dep_sofar) if not self._verify_value( test_dep, ref_dep, unit_aliases, key_dep_sofar): current_error += ( f"- {key_dep_sofar} is {test_dep} but must be " f"{', '.join(ref_dep)} if {key_sofar} is {test}\n") # --------------------------------------------------------------------- # 7. check write only restrictions: units shall be in lower-case # (could be tested in _verify_unit but this way a more verbose error # message can be generated) if mode == "write": keys = [k for k in self.__dict__.keys() if k.endswith("Units")] for key in keys: unit = getattr(self, key) if unit.lower() != unit: current_error += (f"- {key} is {unit} but must contain " "only lower case letters when writing " "SOFA files to disk.\n") if current_error: error_msg += "Detected violations of the SOFA convention:\n" error_msg += current_error # --------------------------------------------------------------------- # 8. check deprecations # (so far there are only deprecations for the convention) if self.GLOBAL_SOFAConventions in \ deprecations["GLOBAL:SOFAConventions"]: msg = ("Detected deprecations:\n" f"- GLOBAL_SOFAConventions is " f"{self.GLOBAL_SOFAConventions}, which is deprecated. Use " "Sofa.upgrade_convention() to upgrade to " f"{deprecations['GLOBAL:SOFAConventions'][self.GLOBAL_SOFAConventions]}") # noqa if mode == "write": error_msg += msg else: warning_msg += msg # warn if preliminary conventions versions are used if float(self.GLOBAL_SOFAConventionsVersion) < 1.0: warning_msg += ( "\n\nDetected preliminary conventions version " f"{self.GLOBAL_SOFAConventionsVersion}:\n - Upgrade data to " "version >= 1.0 if possible. Preliminary conventions might " "change in the future, which could invalidate data that was " "written before the changes.") # --------------------------------------------------------------------- # 9. handle warnings and errors if issue_handling != "ignore": error_occurred, issues = self._verify_handle_issues( warning_msg, error_msg, issue_handling) if error_occurred: if issue_handling == "print": return elif issue_handling == "return": return issues
@staticmethod def _verify_value(test, ref, unit_aliases, key): """ Check a value against the SOFA standard for Sofa.verify() Parameters ---------- test : the value under test ref : the value enforced by the SOFA standard unit_aliases : dict of aliases for units from _verification_rules() key : The name of the current attribute, e.g., "GLOBAL_DataType" Returns ------- ``True`` if `test` and `ref` agree, ``False`` otherwise """ # General case of valid value if ref is None or test in ref: return True # only string data should remain for verification if not isinstance(test, str): raise TypeError(( "At this point, only string data should remain. Please report " "the issue: github.com/pyfar/sofar/issues")) # case sensitive check for DataType and SOFAConventions if key in ["GLOBAL_DataType", "GLOBAL_SOFAConventions"]: if test in ref: return True else: return False # general case insensitive test test = test.lower() if test in ref: return True # if we are not checking a unit, the test value is invalid if key.endswith("Units"): return sf.Sofa._verify_unit(test, ref, unit_aliases) else: return False @staticmethod def _verify_unit(test, ref, unit_aliases): """ Verify if a unit string agrees with AES69 Parameters: ----------- test : string Current unit string (single units or multiple units separated by commas, commas plus spaces, or spaces). ref : list List of length one containing the LOWER reference case unit string, i.e., only the keys from unit_aliases are allowed words. unit_aliases : dict dict of aliases for units from _verification_rules() Returns ------- verified : bool """ # check format of reference units if not isinstance(ref, list) \ or len(ref) != 1 \ or not isinstance(ref[0], str): raise TypeError("ref must be a list of length 1 containing a str") # Following the SOFA standard AES69, units may be separated by # `, ` (comma and space), `,` (comma only), and ` ` (space only). # (regexp ', ?' matches ', ' and ',') units_ref = re.split(', ?', ref[0]) units_test = re.split(', ?', test) # check if number of units agree if len(units_ref) != len(units_test): return False # check if units are valid for unit_test, unit_ref in zip(units_test, units_ref): if unit_test not in unit_aliases \ or unit_aliases[unit_test] != unit_ref: return False return True @staticmethod def _get_reference_unit(test, unit_aliases): """ Return units in reference for, .e.g., "meter" is converted to "metre" and "degrees degrees,meter" is converted to "degree, degree, metre" Parameters: ----------- test : string Current unit string MUST be valid, i.e., tested with Sofa._verify_unit (single units or multiple units separated by commas, commas plus spaces, or spaces). unit_aliases : dict dict of aliases for units from _verification_rules() Returns ------- reference_units : str """ # Following the SOFA standard AES69, units may be separated by # `, ` (comma and space), `,` (comma only), and ` ` (space only). # (regexp ', ?' matches ', ' and ',') units_test = re.split(', ?| ', test.lower()) # get list of reference units units = [unit_aliases[u] for u in units_test] # get reference unit string units = ", ".join(units) if units[0] != "cubic" else " ".join(units) return units @staticmethod def _verify_handle_issues(warning_msg, error_msg, issue_handling): """Handle warnings and errors from Sofa.verify""" # handle warnings if warning_msg != "\nWARNINGS\n--------\n": if issue_handling == "raise": warnings.warn(UserWarning(warning_msg)) elif issue_handling == "print": print(warning_msg) else: warning_msg = None # handle errors if error_msg != "\nERRORS\n------\n": if issue_handling == "raise": raise ValueError(error_msg) elif issue_handling == "print": print(error_msg) else: error_msg = None # flag indicating if an error occurred error_occurred = error_msg is not None # verbose issue message if warning_msg and error_msg: issues = error_msg + "\n" + warning_msg elif warning_msg: issues = warning_msg elif error_msg: issues = error_msg else: issues = None return error_occurred, issues @staticmethod def _verification_rules(): """ Return dictionaries to verify SOFA objects in Sofa.verify(). For detailed information see folder 'sofa_conventions'. Returns: rules : dict All general and specific verification rules unit_aliases : dict Aliases for specific units allowed in SOFA deprecations : dict Deprecated conventions and their substitute upgrade : dict Rules for upgrading deprecated conventions """ base = os.path.join( os.path.dirname(__file__), "sofa_conventions", "rules") with open(os.path.join(base, "rules.json"), "r") as file: rules = json.load(file) with open(os.path.join(base, "unit_aliases.json"), "r") as file: unit_aliases = json.load(file) with open(os.path.join(base, "deprecations.json"), "r") as file: deprecations = json.load(file) with open(os.path.join(base, "upgrade.json"), "r") as file: upgrade = json.load(file) return rules, unit_aliases, deprecations, upgrade
[docs] def copy(self): """Return a copy of the SOFA object.""" return deepcopy(self)
def _reset_convention(self): """ - Add SOFA convention to SOFA object in private attribute `_convention`. If The object already contains a convention, it will be overwritten. - If the SOFA object contains custom entries, check if any of the custom entries part of the convention. If yes, delete the entry from self._custom - If the SOFA objects contains custom entries, add entries from self._custom to self._convention """ # verify convention and version c_current = self.GLOBAL_SOFAConventions v_current = str(self.GLOBAL_SOFAConventionsVersion) _verify_convention_and_version(v_current, c_current) # load and add convention and version convention = self._load_convention( c_current, v_current) self._convention = convention if hasattr(self, "_custom"): # check of custom fields can be removed for key in self._convention: if key in self._custom: del self._custom[key] # check if custom fields can be added for key in self._custom: self._convention[key] = self._custom[key] def _load_convention(self, convention, version): """ Load SOFA convention from json file. Parameters ---------- convention : str The name of the convention from which the SOFA file is created. See :py:func:`~sofar.list_conventions`. version : str ``'latest'`` Use the latest API and upgrade the SOFA file if required. str Version string, e.g., ``'1.0'``. Returns ------- convention : dict The SOFA convention as a dictionary """ # check input if not isinstance(convention, str): raise TypeError(("Convention must be a string " f"but is of type {type(convention)}")) # get and check path to json file paths = _get_conventions("path") path = [path for path in paths if os.path.basename(path).startswith(convention + "_")] if not len(path): raise ValueError( (f"Convention '{convention}' not found. See " "sofar.list_conventions() for available conventions.")) # get available versions as strings versions = [p.split('_')[-1][:-5] for p in path] # select the correct version if version == "latest": versions = np.array([float(v) for v in versions]) path = path[np.argmax(versions)] else: if version not in versions: raise ValueError(( f"Version {version} not found. " f"Available versions are {versions}")) path = path[versions.index(version)] # read convention from json file with open(path, "r") as file: convention = json.load(file) # replace ':' and '.' in key names by '_' convention = { key.replace(':', '_'): value for key, value in convention.items()} convention = { key.replace('.', '_'): value for key, value in convention.items()} return convention def _convention_to_sofa(self, mandatory): """ Use SOFA convention to create attributes with default values. Parameters ---------- mandatory : bool Flag to indicate if only mandatory fields are to be included. """ # populate the SOFA file for key in self._convention.keys(): # skip optional fields if requested if not self._mandatory(self._convention[key]["flags"]) \ and mandatory: continue # get the default value default = self._convention[key]["default"] if isinstance(default, list): ndim = len(self._convention[key]["dimensions"].split(", ")[0]) default = _atleast_nd(default, ndim) # create attribute with default value setattr(self, key, default) # write API and date specific fields (some read only) now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.protected = False self.GLOBAL_DateCreated = now self.GLOBAL_DateModified = now self.GLOBAL_APIName = "sofar SOFA API for Python (pyfar.org)" self.GLOBAL_APIVersion = sf.version() self.GLOBAL_ApplicationName = "Python" self.GLOBAL_ApplicationVersion = ( f"{platform.python_version()} " f"[{platform.python_implementation()} - " f"{platform.python_compiler()}]") self.protected = True @staticmethod def _get_size_and_shape_of_string_var(value, key): """ String variables can be strings, list of strings, or numpy arrays of strings. This functions returns the length of the longest string S inside the string variable and the shape of the string variable as required by the SOFA definition. Note that the shape is the shape of the array that holds the strings. NETCDF stores all string variables in arrays. """ if isinstance(value, str): S = len(value) shape = (1, 1) elif isinstance(value, list): S = len(max(value, key=len)) shape = np.array(value).shape elif isinstance(value, np.ndarray): S = max(np.vectorize(len)(value)) shape = value.shape else: raise TypeError((f"{key} must be a string, numpy string array, " "or list of strings")) return S, shape @staticmethod def _mandatory(flags): """ Check if a field is mandatory Parameters ---------- flags : None, str The flags from convention[key]["flags"] Returns ------- is_mandatory : bool """ # skip optional fields if requested if flags is None: is_mandatory = False elif "m" not in flags: is_mandatory = False else: is_mandatory = True return is_mandatory @staticmethod def _read_only(flags): """ Check if a field is read only Parameters ---------- flags : None, str The flags from convention[key]["flags"] Returns ------- is_read_only : bool """ # skip optional fields if requested if flags is None: is_read_only = False elif "r" not in flags: is_read_only = False else: is_read_only = True return is_read_only