Source code for geogen.product

"""The ``product`` module holds the :class:`Product <geogen.product.Product>` class, the :ref:`instrument_product_class`
inherating from the Product class, and :class:`PLFParser <geogen.product.PLFParser>` and
:class:`ProductFactory <geogen.product.Product>` classes, respectively used to parse an input PLF file and create
corresponding valid Product objects.

.. figure:: ../../images/product_module.png
  :width: 99%
  :align: center

.. _props_dict:

``props_dict``
--------------

A product properties dictionary typically comes from parsing a PLF file. It must contain the following keys:
``instrument_host_id``, ``instrument_id``, ``product_type`` so as to identify and create a corresponding
:ref:`Instrument Product class <instrument_product_class>` object. Additional required keys are defined in the
:ref:`config_file_spec` and retuned by a :meth:`geogen.config.Config.get_product_req_props` method call.
Example of product properties dictionary::

    {
        "product_id": "HH852_0008_SR2.IMG",
        "instrument_host_id": "MEX",
        "instrument_id": "HRSC",
        "product_type": "",
        "target_name": "MARS",
        "start_time": "2018-02-08T02:08:35.476Z",
        "stop_time": "2018-02-08T02:08:35.481Z",
        "detector_id": "MEX_HRSC_SRC",
        "lines": "1018"
    }
"""

import spiceypy as spice
from loguru import logger
import json

[docs]class PLFParser: """Class that parses :ref:`plf_file_spec` files. Holds valid :class:`Product <geogen.product.Product>` objects resulting from a PLF file parsing. Attributes: config (Config): Config object associated to the parser. plf_file (str): Parsed PLF file. n_product_items (int): Number of valid :ref:`product items <plf_product_json_object>` in PFL file. products (List[Product]): List of Product objects matching valid :ref:`product items <plf_product_json_object>`. """ def __init__(self, config): """Constructs PLFParser object. Args: config (Config): Config object to be associated to the parser. """ self.config = config self.plf_file = '' self.n_product_items = 0 self.products = []
[docs] def parse(self, plf_file): """Parses an input PLF file and returns corresponding Product objects. Args: plf_file (str): PLF file to be parsed. Returns: List[Product]: """ self.plf_file = plf_file with open(plf_file, 'r') as fp: obj = json.load(fp) props_dicts = obj['products'] self.n_product_items = len(props_dicts) product_factory = ProductFactory(self.config) for props_dict in props_dicts: product = product_factory.create(props_dict) if product.valid: self.products.append(product) return self.products
[docs] def get_missions(self): """Returns the list of mission IDs corresponding to valid Product objects. Returns: List[str]: """ missions = [] for product in self.products: if product.instrument_host_id not in missions: missions.append(product.instrument_host_id) return missions
[docs]class ProductFactory: """Class that creates Product or Product-child objects. Attributes: config (Config): Associated Config object. """ def __init__(self, config): """Constructs ProductFactory object. Args: config (Config): Input Config object. """ self.config = config
[docs] def create(self, props_dict, silent=False): """Returns a valid Product-children class object from an input product props dictionary. Args: props_dict (dict): Input product properties dictionary (see :ref:`props_dict`). silent (bool, optional): Indicates whether or not missing-required-product-property warning message is kept silent. Returns: Product: """ # Set input product ID product_id = 'UNKNOWN' if 'product_id' in props_dict.keys(): product_id = props_dict['product_id'] # Set basic required properties that must be in input props dictionary if 'instrument_host_id' in props_dict.keys(): instrument_host_id = props_dict['instrument_host_id'] # Check and update input instrument host ID to allowed configuration mission key. naif_instrhost_code, found = spice.bodn2c(instrument_host_id) if found: naif_instrhost_name, found = spice.bodc2n(naif_instrhost_code) instrument_host_id = naif_instrhost_name props_dict['instrument_host_id'] = naif_instrhost_name else: logger.error(f'SPICE-unknown input <{instrument_host_id}> instrument host ID for <{product_id}> product.') return Product({}, self.config, silent=True) else: return None if 'instrument_id' in props_dict.keys(): instrument_id = props_dict['instrument_id'] else: return Product({}, self.config, silent=True) if 'product_type' in props_dict.keys(): product_type = props_dict['product_type'] if product_type is None: # change None to empty string product_type = '' props_dict['product_type'] = '' else: return Product({}, self.config, silent=True) # Get product class name for given instrument_host_id, instrument_id and product_type product_class_name = self.config.get_product_class(instrument_host_id, instrument_id, product_type) if product_class_name != '': product_class = globals()[product_class_name] product = product_class(props_dict, self.config, silent=silent) else: logger.warning('') logger.warning("Invalid input product props for <{}>: no <{}> product type definition for <{}> instrument " "host and <{}> instrument.".format(product_id, product_type, instrument_host_id, instrument_id)) logger.warning('-> You can add new product type definition in config file: {}'.format(self.config.config_file)) logger.warning('') product = Product(props_dict, self.config) product.valid = False return product
[docs]class Product: """Class that represents any PDS observational data product. Its hold the observational data product properties required to initiate a valid :class:`Observation <geogen.coverage.Observation>` object, and compute its geometry metadata. Attributes: instrument_host_id (str): Instrument host ID. instrument_id (str): Instrument ID. product_type (str): Product type. detector_name (str): Detector name. detector_mode (str): Detector mode (default is 1). n_data_records (int): Number of data records (default is 0). detector_subframe (bool): Boolean defining whether a FRAME detector can have sub-frames. detector_fov_ref_ratio (float): Detector sub-frame FOV reference-axis ratio. detector_fov_cross_ratio (float): Detector sub-frame FOV cross-axis ratio. time_offset (float): Time offset in seconds to be applied for computation. config (Config): Associated Config object. valid (bol): Whether or not Product object is valid. """ def __init__(self, props_dict, config, silent=False): """Constructs Product object. Args: props_dict (dict): Input product properties dictionary (see :ref:`props_dict`). config (Config): Input Config object. silent (bool): Indicates whether or not missing-required-product-property warning message is kept silent. """ self.instrument_host_id = '' self.instrument_id = '' self.product_type = '' self.detector_name = '' self.detector_mode = 1 self.n_data_records = 0 self.detector_subframe = False self.detector_fov_ref_ratio = 1.0 self.detector_fov_cross_ratio = 1.0 self.time_offset = 0.0 self.config = config self.valid = True # Set basics input properties that are required to retrieve all required properties. # if 'instrument_host_id' in props_dict.keys(): self.instrument_host_id = props_dict['instrument_host_id'] if 'instrument_id' in props_dict.keys(): self.instrument_id = props_dict['instrument_id'] if 'product_type' in props_dict.keys(): self.product_type = props_dict['product_type'] product_id = 'UNKNOWN' if 'product_id' in props_dict.keys(): product_id = props_dict["product_id"] # Get all required product properties for a given instrument host, instrument and product type. # req_props_names, req_props_paths, default_product_type = \ self.config.get_product_req_props(self.instrument_host_id, self.instrument_id, self.product_type) # Check that all required properties are available from input dict and are valid # for req_props_name in req_props_names: if not req_props_name in props_dict.keys(): self.valid = False if not silent: logger.warning('Missing <{}> required product property for <{}>.'.format(req_props_name, product_id)) else: # Set required valid attribute to instantiated Product object setattr(self, req_props_name, props_dict[req_props_name]) if not self.valid: return # Add not-required properties with a warning (any product properties can be passed to output coverage # GeoJSON geometry properties). # for prop in props_dict.keys(): if not prop in req_props_names: setattr(self, prop, props_dict[prop]) #logger.warning('Input <{}> property is not required'.format(prop)) if self.product_type == '': self.product_type = default_product_type # Attempt to assign detector name if self.valid: self.detector_name = self.instrument_host_id + '_' + self.instrument_id def __repr__(self): return ( f"<{self.__class__.__name__}> " f"Instrument Host ID: {self.instrument_host_id} | " f"Instrument ID: {self.instrument_id} | " f"Product Type: {self.product_type} | " f"Detector Name: {self.detector_name} | " f"Detector Mode: {self.detector_mode} | " f"Valid: {self.valid}" )
# ====================================================================================================================== # Product-child classes # ====================================================================================================================== # # Template for a new instrument Product child class:: # # class <MISSION_INSTRUMENT_TYPE>(Product): # def __init__(self, pdict, config): # super().__init__(pdict, config) # # if self.valid: # # insert code ... # MEX Product classes --------------------------------------------------------------------------------------------------
[docs]class MEX_OMEGA_EDR(Product): """Product class for MEX/OMEGA EDR data products. - ``detector_name`` is always set to 'MEX_OMEGA_SWIR_C'. - ``detector_mode`` is set to the first element of the data cube dimension, given by the ``core_items`` property value. - ``n_data_records`` is set to the third element of the data cube dimension, given by the ``core_items`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'MEX_OMEGA_SWIR_C' self.detector_mode = int(self.core_items[1:-1].split(',')[0]) self.n_data_records = int(self.core_items[1:-1].split(',')[2])
[docs]class MEX_HRSC_RDR(Product): """Product class for MEX/HRSC RDR data products. - ``detector_name`` is set to the ``detector_id`` property value. - ``n_data_records`` is set to the ``lines`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = self.detector_id self.n_data_records = int(self.lines)
[docs]class MEX_PFS_EDR(Product): """Product class for MEX/PFS EDR data products. - ``detector_name`` is set based on the ``detector_id`` property value. - ``n_data_records`` is set to the ``file_records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'MEX_PFS_' + self.detector_id + 'C' self.n_data_records = int(self.file_records)
[docs]class MEX_SPICAM_RDR(Product): """Product class for MEX/SPICAM RDR data products. - ``detector_name`` is set based on the ``channel_id``, ``target_type``, ``spacecraft_pointing_mode`` and ``spicam_binning`` properties values. - ``n_data_records`` is set to the ``file_records`` property value. - ``target_name`` is set to the ``spicam_target`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: # Derive detector name # # MEX_SPICAM_UV_NADIR_BIN_32 -> MEX_SPICAM_UV1_NADIR_BIN_32 # MEX_SPICAM_UV_STELLAR -> MEX_SPICAM_UV1_STAR # MEX_SPICAM_UV_SOLAR -> MEX_SPICAM_UV2_SUN # MEX_SPICAM_IR_SOLAR -> MEX_SPICAM_IR2_SUN # detector_name = 'MEX_SPICAM_' + self.channel_id + '1_' if (self.target_type == 'STAR') or (self.target_type == 'CALIBRATION') or (self.target_type == 'N/A'): detector_name += 'STAR' elif (self.target_type == 'PLANET') or (self.target_type == 'SATELLITE'): if (self.spacecraft_pointing_mode) == 'NADIR' or (self.spacecraft_pointing_mode == 'INERT'): detector_name += 'NADIR_BIN_' + str(self.spicam_binning) elif (self.target_type == 'SUN'): detector_name += 'SUN' if detector_name == 'MEX_SPICAM_UV1_SUN': detector_name = 'MEX_SPICAM_UV2_SUN' if detector_name == 'MEX_SPICAM_IR1_SUN': detector_name = 'MEX_SPICAM_IR2_SUN' self.detector_name = detector_name self.n_data_records = int(self.file_records) self.target_name = self.spicam_target
[docs]class MEX_MARSIS(Product): r"""Product class for MEX/MARSIS data products. Currently defined as applicable to EDR and RDR product types in `psa_config.json`_ file. Temporary detector FOV definition in `mex_addendum.ti`_ file:: \begindata INS-41300_FOV_FRAME = 'MEX_SPACECRAFT' INS-41300_FOV_SHAPE = 'RECTANGLE' INS-41300_BORESIGHT = ( 0.00000 0.00000 1.00000 ) INS-41300_FOV_CLASS_SPEC = 'ANGLES' INS-41300_FOV_REF_VECTOR = ( 1.00000 0.00000 0.00000 ) INS-41300_FOV_REF_ANGLE = 1.0 INS-41300_FOV_CROSS_ANGLE = 1.0 INS-41300_FOV_ANGLE_UNITS = 'DEGREES' \begintext - ``detector_name`` is always set to 'MEX_MARSIS' - ``n_data_records`` is set to the ``rows`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'MEX_MARSIS' self.n_data_records = int(self.rows)
[docs]class MEX_ASPERA(Product): """Product class for MEX/ASPERA-3 data products. Currently defined as applicable to DATA and DDR product types in `psa_config.json`_ file. - ``detector_name`` is set based on ``product_id`` property value. - ``n_data_records`` is set to ``file_records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: sub_instr = self.product_id[:3] # eg: "IMA_AZ0020072640742C_ACCS01" -> IMA self.detector_name = 'MEX_ASPERA_' + sub_instr self.n_data_records = int(self.file_records)
[docs]class MEX_VMC(Product): """Product class for MEX/VMC data products. Currently defined as applicable to EDR and RDR product types in `psa_config.json`_ file. - ``detector_name`` is set to ``detector_id`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = self.detector_id
# ROSETTA Product classes ----------------------------------------------------------------------------------------------
[docs]class ROS_OSIRIS(Product): """Product class for ROSETTA/OSIRIS WAC and NAC data products. Currently defined as applicable to EDR product type in `psa_config.json`_ file. - ``detector_name`` is set based on ``instrument_id`` and ``geometric_distortion_correction_flag`` properties values. - ``target_name`` is currently updated so as to take into account for undefined SPICE body name synonyms. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: # TODO: this shouldn't needed anymore since target assigment method has been implemented if (self.target_name == '67P/C-G') or (self.target_name == '67P/CHURYUMOV-GERASIMENKO') or \ (self.target_name == '67P/CHURYUMOV-GERASIMENKO 1 (1969 R1)') or \ (self.target_name == '67P/CHURYUMOV-GERASIMENKO 1969 R1'): self.target_name = '67P/CHURYUMOV-GERASIMENKO' dist = '' sub_instr = '' if self.instrument_id == 'OSINAC': sub_instr = 'NAC' elif self.instrument_id == 'OSIWAC': sub_instr = 'WAC' if self.geometric_distortion_correction_flag: dist = '_DIST' self.detector_name = 'ROS_OSIRIS_' + sub_instr + dist
[docs]class ROS_NAVCAM_EDR(Product): """Product class for ROSETTA/NAVCAM EDR data products. - ``detector_name`` is set based on ``channel_id`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: channel = '' if self.channel_id == 'CAM1': channel = 'A' elif self.channel_id == 'CAM2': channel = 'B' else: logger.warning('Unknown NAVCAM Channel ID: ' + self.channel_id ) self.detector_name = 'ROS_NAVCAM-' + channel
[docs]class ROS_ALICE(Product): """Product class for ROSETTA/ALICE data products. Currently defined as applicable to EDR, REDR and REFDR product types in `psa_config.json`_ file. - ``detector_name`` is always set to 'ROS_ALICE'. - ``n_data_records`` is set to ``lines`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'ROS_ALICE' self.n_data_records = int(self.lines)
[docs]class ROS_VIRTIS_EDR(Product): """Product class for ROSETTA/VIRTIS EDR data products. - ``detector_name`` is set based on ``channel_id`` property value. - ``n_data_records`` is set to the third element of the data cube dimension, given by the ``core_items`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.n_data_records = int(self.core_items[1:-1].split(',')[2]) if self.channel_id == 'VIRTIS_M': self.detector_name = 'ROS_VIRTIS-M' elif self.channel_id == 'VIRTIS_M_VIS': self.detector_name = 'ROS_VIRTIS-M_VIS_ZERO' elif self.channel_id == 'VIRTIS_M_VIS_ZERO': self.detector_name = 'ROS_VIRTIS-M_VIS_ZERO' elif self.channel_id == 'VIRTIS_M_IR': self.detector_name = 'ROS_VIRTIS-M_IR' elif self.channel_id == 'VIRTIS_M_IR_ZERO': self.detector_name = 'ROS_VIRTIS-M_IR_ZERO' elif self.channel_id == 'VIRTIS_H': self.detector_name = 'ROS_VIRTIS-H' else: logger.warning('Unknown VIRTIS Channel ID :' + self.channel_id )
[docs]class ROS_MIRO(Product): """Product class for ROSETTA/MIRO data products. Currently defined as applicable to EDR and RDR product types in `psa_config.json`_ file. - ``detector_name`` is set based on ``product_id`` property value. - ``n_data_records`` is set to ``file_records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: # Derive detector name if ("_MM_" in self.product_id) or \ ("_MMCAL_" in self.product_id) or \ ("_MM1S_" in self.product_id) or \ ("_MMGEOM_" in self.product_id): self.detector_name = 'ROS_MIRO_MM' elif ("_SUBMM_" in self.product_id) or \ ("_CTS_" in self.product_id) or \ ("_SUBMMCAL_" in self.product_id) \ or ("_CTSCAL_" in self.product_id) or \ ("_CTSFOLDED_" in self.product_id) or \ ("_SUBMM1S_" in self.product_id) or \ ("_CTSGEOM_" in self.product_id) or \ ("_SUBMMGEOM_" in self.product_id) or \ ("_SUB_GEOM_" in self.product_id): self.detector_name = 'ROS_MIRO_SUBMM' else: logger.warning('Unknown MIRO detector field in product name :' + self.product_id ) self.n_data_records = int(self.file_records)
# TGO Product classes --------------------------------------------------------------------------------------------------
[docs]class TGO_ACS_RAW(Product): """Product class for TGO/ACS Raw data products. - ``detector_name`` is set based on ``identifier`` property value. - ``n_data_records`` is set to ``records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: channel = self.identifier # Derive detector name self.detector_name = 'TGO_ACS_' + channel # default if channel == 'MIR': self.detector_name = 'TGO_ACS_MIR' elif channel == 'TIR': self.detector_name = 'TGO_ACS_TIRVIM' else: logger.warning('Unable to derive detector name for <{}> product'.format(props_dict['product_id'])) self.n_data_records = int(self.records)
[docs]class TGO_ACS_CAL(Product): """Product class for TGO/ACS Calibrated data products. - ``detector_name`` is based on ``identifier`` and ``nir_observation_type`` properties values. - ``n_data_records`` is set to ``records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.n_data_records = int(self.records) channel = self.identifier # instead of self.product_id.split((':')[-1])[5].split('_')[3].upper() if channel == 'MIR': self.detector_name = 'TGO_ACS_MIR' elif channel == 'NIR': if self.nir_observation_type == 'SO': self.detector_name = 'TGO_ACS_NIR_OCC' elif self.nir_observation_type == 'Nadir': self.detector_name = 'TGO_ACS_NIR_NAD' elif channel == 'TIR': self.detector_name = 'TGO_ACS_TIRVIM' else: logger.warning('Unable to derive detector name for <{}> product'.format(props_dict['product_id']))
[docs]class TGO_CASSIS_RAW(Product): """Product class for TGO/CASSIS Raw data products. - ``detector_name`` is set based on ``product_id`` property value. - ``detector_fov_ref_ratio`` is set based on ``samples`` property value. - ``detector_fov_cross_ratio`` is set based on ``lines`` and ``filter`` properties values. - ``stop_time`` is set to ``start_time`` property value. - No ``time_offset`` is applied (can be changed if deemed necessary). """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: # Set detector name from product_id property # eg: cas_raw_sc_20191101t011741-20191101t011745-8650-69-pan-554586220-0-0 -> pan filter = self.product_id.split(':')[-1].split('-')[4].upper() self.detector_name = 'TGO_CASSIS_' + filter # Set detector sub-frame parameters max_samples = 2048 max_lines = 280 if filter == 'PAN' else 256 self.detector_fov_ref_ratio = self.samples / max_samples self.detector_fov_cross_ratio = self.lines / max_lines # Overwrite stop_time to start_time value self.stop_time = self.start_time
#self.time_offset = 2.88
[docs]class TGO_CASSIS_CAL(Product): """Product class for TGO/CASSIS Calibrated data products. - ``detector_name`` is set based on ``filter`` property value. - ``detector_fov_ref_ratio`` is set based on ``samples`` property value. - ``detector_fov_cross_ratio`` is set based on ``lines`` and ``filter`` properties values. - ``stop_time`` is set to ``start_time`` property value. - No ``time_offset`` is applied (can be changed if deemed necessary). """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: # Set detector name from `filter` property self.detector_name = 'TGO_CASSIS_' + self.filter # Set detector sub-frame parameters max_samples = 2048 max_lines = 280 if self.filter == 'PAN' else 256 self.detector_fov_ref_ratio = self.samples / max_samples self.detector_fov_cross_ratio = self.lines / max_lines # Overwrite stop_time to start_time value self.stop_time = self.start_time
#self.time_offset = 2.88
[docs]class TGO_FREND_CAL(Product): """Product class for TGO/FREND Calibrated data products. - ``detector_name`` is always set to 'TGO_FREND'. - ``n_data_records`` is set to ``records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'TGO_FREND' self.n_data_records = int(self.records)
[docs]class TGO_NOMAD_RAW(Product): """Product class for TGO/NOMAD Raw data products. - ``detector_name`` is set based ``identifier`` property value. - ``n_data_records`` is set to ``records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.detector_name = 'TGO_NOMAD_' + self.identifier self.n_data_records = int(self.records)
[docs]class TGO_NOMAD_CAL(Product): """Product class for TGO/NOMAD Calibrated data products. - ``detector_name`` is set based on ``identifier`` and ``product_id`` properties values. - ``n_data_records`` is set to ``records`` property value. """ def __init__(self, props_dict, config, silent=False): super().__init__(props_dict, config, silent=silent) if self.valid: self.n_data_records = int(self.records) # Defining UVIS Occultation and Nadir channel = self.identifier # instead of self.product_id.split((':')[-1])[5].split('_')[3].upper() self.detector_name = 'TGO_NOMAD_' + channel if channel == 'UVIS': obs_type = self.product_id.split((':')[-1])[5].split('-')[2] if obs_type == 'd': self.detector_name = 'TGO_NOMAD_' + channel + '_NAD' if obs_type == 'e' or obs_type == 'i': self.detector_name = 'TGO_NOMAD_' + channel + '_OCC'