mt_metadata.timeseries.run ========================== .. py:module:: mt_metadata.timeseries.run Classes ------- .. autoapisummary:: mt_metadata.timeseries.run.Run Module Contents --------------- .. py:class:: Run(**data) Bases: :py:obj:`mt_metadata.base.MetadataBase` Base class for all metadata objects with Pydantic validation. MetadataBase extends DotNotationBaseModel (which inherits from Pydantic's BaseModel) to provide automatic validation according to metadata standards. It adds functionality beyond dictionaries, supporting JSON, XML, pandas Series, and other formats for metadata interchange. .. attribute:: _skip_equals Private attribute listing fields to skip in equality comparisons :type: list[str] .. attribute:: _fields Private attribute caching field information :type: dict[str, Any] .. rubric:: Notes - All field assignments are validated automatically via Pydantic - None values are converted to appropriate defaults (empty string or 0.0) - Supports nested attribute access via dot notation - Thread-safe for read operations after initialization .. py:attribute:: channels_recorded_auxiliary :type: Annotated[list[str], Field(default_factory=list, description='List of auxiliary channels recorded', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['[T]']})] .. py:attribute:: channels_recorded_electric :type: Annotated[list[str], Field(default_factory=list, description='List of electric channels recorded', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['[Ex , Ey]']})] .. py:attribute:: channels_recorded_magnetic :type: Annotated[list[str], Field(default_factory=list, description='List of magnetic channels recorded', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['[Hx , Hy , Hz]']})] .. py:method:: channels_recorded_all() List of all channels recorded in the run. .. py:attribute:: comments :type: Annotated[mt_metadata.common.Comment, Field(default_factory=Comment, description='Any comments on the run.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['cows chewed cables']})] .. py:attribute:: data_type :type: Annotated[mt_metadata.common.DataTypeEnum, Field(default=DataTypeEnum.BBMT, description='Type of data recorded for this run.', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['BBMT']})] .. py:attribute:: id :type: Annotated[str, Field(default='', description='Run ID should be station name followed by a number or character. Characters should only be used if the run number is small, if the run number is high consider using digits with zeros. For example if you have 100 runs the run ID could be 001 or {station}001.', alias=None, pattern='^[a-zA-Z0-9_]*$', json_schema_extra={'units': None, 'required': True, 'examples': ['001']})] .. py:attribute:: sample_rate :type: Annotated[float, Field(default=0.0, description='Digital sample rate for the run', alias=None, json_schema_extra={'units': 'samples per second', 'required': True, 'examples': ['100']})] .. py:attribute:: acquired_by :type: Annotated[mt_metadata.common.AuthorPerson, Field(default_factory=AuthorPerson, description='Information about the group that collected the data.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Person()']})] .. py:attribute:: metadata_by :type: Annotated[mt_metadata.common.AuthorPerson, Field(default_factory=AuthorPerson, description='Information about the group that collected the metadata.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Person()']})] .. py:attribute:: provenance :type: Annotated[mt_metadata.common.Provenance, Field(default_factory=Provenance, description='Provenance information about the run.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Provenance()']})] .. py:attribute:: time_period :type: Annotated[mt_metadata.common.TimePeriod, Field(default_factory=TimePeriod, description='Time period for the run.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ["TimePeriod(start='2020-01-01', end='2020-12-31')"]})] .. py:attribute:: data_logger :type: Annotated[mt_metadata.timeseries.DataLogger, Field(default_factory=DataLogger, description='Data Logger information used to collect the run.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['DataLogger()']})] .. py:attribute:: fdsn :type: Annotated[mt_metadata.common.Fdsn, Field(default_factory=Fdsn, description='FDSN information for the run.', alias=None, json_schema_extra={'units': None, 'required': False, 'examples': ['Fdsn()']})] .. py:attribute:: channels :type: Annotated[mt_metadata.common.list_dict.ListDict | list | dict | collections.OrderedDict, Field(default_factory=ListDict, description='ListDict of channel objects collected in this run.', alias=None, exclude=True, json_schema_extra={'units': None, 'required': False, 'examples': ['ListDict(Electric(), Magnetic(), Auxiliary())']})] .. py:method:: validate_comments(value, info) :classmethod: Validate that the value is a valid comment. .. py:method:: validate_data_type(value, info) :classmethod: Validate that the data_type is a string. .. py:method:: validate_list_of_strings(value, info) :classmethod: Validate that the value is a list of strings. .. py:method:: validate_channels_recorded() Validate that the value is a list of strings. .. py:method:: validate_channels(value, info) :classmethod: .. py:method:: merge(other, inplace = True) Merge channels from another Run into this run. Combines channels from two runs and updates the channels recorded lists and time period. :param other: Another Run object whose channels will be merged into this run. :type other: Run :param inplace: If True, update this run and update time period. If False, return a copy of the merged run (default is True). :type inplace: bool, optional :returns: If inplace is False, returns a copy of the merged Run. Otherwise None. :rtype: Run | None :raises TypeError: If other is not a Run object. .. rubric:: Examples Merge runs in place: >>> run1 = Run(id='001') >>> run1.add_channel(Electric(component='ex')) >>> run2 = Run(id='002') >>> run2.add_channel(Magnetic(component='hx')) >>> run1.merge(run2, inplace=True) >>> print(run1.channels_recorded_all) ['ex', 'hx'] Merge and return new run: >>> merged_run = run1.merge(run2, inplace=False) >>> print(merged_run.channels_recorded_all) ['ex', 'hx'] .. seealso:: :py:obj:`update` Update metadata from another run .. py:method:: update(other, match = []) Update attribute values from another Run object. Copies non-None, non-default attribute values from another Run object to this one. Skips empty values like None, 0.0, [], empty strings, and default timestamps. :param other: Another Run object to copy attributes from. :type other: Run :param match: List of attribute names that must match between runs before updating. If any don't match, raises ValueError. Typically used for 'id' to ensure runs are compatible (default is None). :type match: list[str] | None, optional :raises ValueError: If any attributes in match list don't have equal values. :raises TypeError: If other is not a compatible Run type. .. rubric:: Examples Basic update: >>> run1 = Run(id='001', sample_rate=256.0) >>> run2 = Run(id='001', sample_rate=0.0) >>> run2.acquired_by.author = 'J. Doe' >>> run1.update(run2) >>> print(run1.acquired_by.author) 'J. Doe' >>> print(run1.sample_rate) # Not updated (run2 has default 0.0) 256.0 Update with matching check: >>> run1 = Run(id='001') >>> run2 = Run(id='002') >>> try: ... run1.update(run2, match=['id']) ... except ValueError as e: ... print("IDs don't match!") IDs don't match! .. rubric:: Notes Channel metadata is also updated. For each channel in other, if the channel exists in this run, it's updated; if not, it's added. Skipped values: - None - 0.0 - Empty lists [] - Empty strings '' - Default timestamp '1980-01-01T00:00:00+00:00' .. seealso:: :py:obj:`merge` Merge channels from another run .. py:method:: has_channel(component) Check if a channel with the given component exists in the run. :param component: Channel component name to search for (e.g., 'ex', 'hy'). :type component: str :returns: True if channel exists, False otherwise. :rtype: bool .. rubric:: Examples >>> run = Run(id='001') >>> run.add_channel(Electric(component='ex')) >>> print(run.has_channel('ex')) True >>> print(run.has_channel('ey')) False .. seealso:: :py:obj:`get_channel` Retrieve a channel object :py:obj:`channel_index` Get the index of a channel .. py:method:: channel_index(component) Get the index of a channel in the channels_recorded_all list. :param component: Channel component name to search for (e.g., 'ex', 'hy'). :type component: str :returns: Index of the channel if found, None otherwise. :rtype: int | None .. rubric:: Examples >>> run = Run(id='001') >>> run.add_channel(Electric(component='ex')) >>> run.add_channel(Electric(component='ey')) >>> run.add_channel(Magnetic(component='hx')) >>> print(run.channel_index('ey')) 1 >>> print(run.channel_index('hz')) None .. rubric:: Notes Channels are sorted alphabetically in channels_recorded_all. .. seealso:: :py:obj:`has_channel` Check if channel exists :py:obj:`get_channel` Retrieve channel object .. py:method:: get_channel(component) Retrieve a channel object by component name. :param component: Channel component name to retrieve (e.g., 'ex', 'hy'). :type component: str :returns: Channel object if found, None otherwise. Return type depends on the channel type. :rtype: Electric | Magnetic | Auxiliary | None .. rubric:: Examples >>> run = Run(id='001') >>> ex = Electric(component='ex', dipole_length=100.0) >>> run.add_channel(ex) >>> channel = run.get_channel('ex') >>> print(type(channel).__name__) 'Electric' >>> print(channel.dipole_length) 100.0 >>> print(run.get_channel('ey')) None .. seealso:: :py:obj:`has_channel` Check if channel exists :py:obj:`add_channel` Add a channel to the run .. py:method:: add_channel(channel_obj, update = True) Add or update a channel in the run. If the channel already exists (matched by component), its metadata is updated. If it doesn't exist, it's added to the channels list. Can accept channel objects, dictionaries, or component strings. :param channel_obj: Channel to add. Can be: - Channel object (Electric, Magnetic, or Auxiliary) - Dictionary with channel attributes (must include 'type' or 'component') - String component name (e.g., 'ex', 'hy', 'temp') If string, channel type is inferred: - Starts with 'e' → Electric - Starts with 'h' or 'b' or equals 'magnetic' → Magnetic - Otherwise → Auxiliary :type channel_obj: Electric | Magnetic | Auxiliary | dict | str :param update: If True, update the run's time period to include this channel's time period. If False, don't update time period (default is True). :type update: bool, optional .. rubric:: Examples Add channel objects: >>> run = Run(id='001') >>> ex = Electric(component='ex', dipole_length=100.0) >>> run.add_channel(ex) >>> print(run.channels_recorded_electric) ['ex'] Add from string (infers type): >>> run.add_channel('hy') >>> run.add_channel('temperature') >>> print(run.channels_recorded_magnetic) ['hy'] >>> print(run.channels_recorded_auxiliary) ['temperature'] Add from dictionary: >>> channel_dict = { ... 'type': 'electric', ... 'component': 'ey', ... 'dipole_length': 95.0 ... } >>> run.add_channel(channel_dict) Update existing channel: >>> ex_updated = Electric(component='ex', dipole_length=105.0) >>> run.add_channel(ex_updated) # Updates existing 'ex' >>> print(run.get_channel('ex').dipole_length) 105.0 Add without updating time period: >>> run.add_channel('hz', update=False) .. rubric:: Notes This method automatically: - Updates channels_recorded lists - Updates run time period (if update=True) - Converts string/dict inputs to proper channel objects - Logs when updating existing channels .. seealso:: :py:obj:`remove_channel` Remove a channel from the run :py:obj:`get_channel` Retrieve a channel object :py:obj:`update_time_period` Manually update time period .. py:method:: remove_channel(channel_id) Remove a channel from the run. :param channel_id: Channel component name to remove (e.g., 'ex', 'hy'). :type channel_id: str .. rubric:: Examples >>> run = Run(id='001') >>> run.add_channel(Electric(component='ex')) >>> run.add_channel(Electric(component='ey')) >>> print(run.channels_recorded_electric) ['ex', 'ey'] >>> run.remove_channel('ex') >>> print(run.channels_recorded_electric) ['ey'] >>> run.remove_channel('ez') # Doesn't exist # Logs warning: Could not find ez to remove. .. rubric:: Notes Automatically updates the channels_recorded lists after removal. Logs a warning if the channel is not found. .. seealso:: :py:obj:`add_channel` Add a channel to the run :py:obj:`has_channel` Check if channel exists .. py:method:: update_channel_keys() Update channel dictionary keys to match current component values. Updates the keys in the channels ListDict to match current channel components. Useful when channel components have been modified after channels were added, ensuring channels can be accessed by their current component values. :returns: Mapping of old keys to new keys showing what was changed. :rtype: dict[str, str] .. rubric:: Examples Fix keys after modifying components: >>> run = Run(id='001') >>> channel = Electric(component='') >>> run.add_channel(channel) >>> # Channel is stored with empty string key >>> channel.component = 'ex' >>> key_mapping = run.update_channel_keys() >>> print(key_mapping) {'': 'ex'} >>> # Now accessible as run.channels['ex'] >>> print(run.get_channel('ex').component) 'ex' Multiple key updates: >>> run = Run(id='001') >>> ch1 = Electric(component='e1') >>> ch2 = Magnetic(component='h1') >>> run.add_channel(ch1) >>> run.add_channel(ch2) >>> ch1.component = 'ex' >>> ch2.component = 'hx' >>> mapping = run.update_channel_keys() >>> print(mapping) {'e1': 'ex', 'h1': 'hx'} .. rubric:: Notes This is typically only needed if you've directly modified channel component attributes after adding them to the run. Normal usage doesn't require calling this method. .. seealso:: :py:obj:`add_channel` Add channels to the run :py:obj:`get_channel` Access channels by component .. py:property:: n_channels :type: int Number of channels in the run. :returns: Count of channels currently in the run. :rtype: int .. rubric:: Examples >>> run = Run(id='001') >>> print(run.n_channels) 0 >>> run.add_channel('ex') >>> run.add_channel('hy') >>> print(run.n_channels) 2 .. py:method:: update_time_period() Update run's time period to encompass all channel time periods. Examines all channels in the run and updates the run's start and end times to include the earliest start and latest end from all channels. Ignores default timestamp '1980-01-01T00:00:00+00:00'. .. rubric:: Examples >>> from mt_metadata.timeseries import Run, Electric >>> run = Run(id='001') >>> ex = Electric(component='ex') >>> ex.time_period.start = '2020-01-01T00:00:00+00:00' >>> ex.time_period.end = '2020-01-01T01:00:00+00:00' >>> run.add_channel(ex, update=False) >>> print(run.time_period.start) 1980-01-01T00:00:00+00:00 >>> run.update_time_period() >>> print(run.time_period.start) 2020-01-01T00:00:00+00:00 Multiple channels: >>> ey = Electric(component='ey') >>> ey.time_period.start = '2020-01-01T00:30:00+00:00' >>> ey.time_period.end = '2020-01-01T02:00:00+00:00' >>> run.add_channel(ey, update=True) >>> print(run.time_period.start) # Uses earliest 2020-01-01T00:00:00+00:00 >>> print(run.time_period.end) # Uses latest 2020-01-01T02:00:00+00:00 .. rubric:: Notes - Only updates if channels exist (n_channels > 0) - Ignores channels with default timestamp - Always expands time period, never shrinks it - Automatically called by add_channel() when update=True .. seealso:: :py:obj:`add_channel` Add channel and optionally update time period