mt_metadata.timeseries.run

Classes

Run

Base class for all metadata objects with Pydantic validation.

Module Contents

class mt_metadata.timeseries.run.Run(**data)

Bases: 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.

_skip_equals

Private attribute listing fields to skip in equality comparisons

Type:: list[str]

_fields

Private attribute caching field information

Type:: dict[str, Any]

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

channels_recorded_auxiliary: Annotated[list[str], Field(default_factory=list, description='List of auxiliary channels recorded', alias=None, json_schema_extra={'units': None, 'required': True, 'examples': ['[T]']})]

channels_recorded_electric: 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]']})]

channels_recorded_magnetic: 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]']})]

channels_recorded_all()

List of all channels recorded in the run.

comments: 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']})]

data_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']})]

id: 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']})]

sample_rate: 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']})]

acquired_by: 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()']})]

metadata_by: 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()']})]

provenance: 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()']})]

time_period: 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')"]})]

data_logger: 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()']})]

fdsn: 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()']})]

channels: 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())']})]

classmethod validate_comments(value, info)

Validate that the value is a valid comment.

classmethod validate_data_type(value, info)

Validate that the data_type is a string.

classmethod validate_list_of_strings(value, info)

Validate that the value is a list of strings.

validate_channels_recorded()

Validate that the value is a list of strings.

classmethod validate_channels(value, info)

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.

Parameters:

other (Run) – Another Run object whose channels will be merged into this run.
inplace (bool, optional) – If True, update this run and update time period. If False, return a copy of the merged run (default is True).

Returns:

If inplace is False, returns a copy of the merged Run. Otherwise None.

Return type:

Run | None

Raises:

TypeError – If other is not a Run object.

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']

See also

update: Update metadata from another run

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.

Parameters:

other (Run) – Another Run object to copy attributes from.
match (list[str] | None, optional) – 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).

Raises:

ValueError – If any attributes in match list don’t have equal values.
TypeError – If other is not a compatible Run type.

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!

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’

See also

merge: Merge channels from another run

has_channel(component)

Check if a channel with the given component exists in the run.

Parameters:: component (str) – Channel component name to search for (e.g., ‘ex’, ‘hy’).
Returns:: True if channel exists, False otherwise.
Return type:: bool

Examples

>>> run = Run(id='001')
>>> run.add_channel(Electric(component='ex'))
>>> print(run.has_channel('ex'))
True
>>> print(run.has_channel('ey'))
False

See also

get_channel: Retrieve a channel object
channel_index: Get the index of a channel

channel_index(component)

Get the index of a channel in the channels_recorded_all list.

Parameters:: component (str) – Channel component name to search for (e.g., ‘ex’, ‘hy’).
Returns:: Index of the channel if found, None otherwise.
Return type:: int | None

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

Notes

Channels are sorted alphabetically in channels_recorded_all.

See also

has_channel: Check if channel exists
get_channel: Retrieve channel object

get_channel(component)

Retrieve a channel object by component name.

Parameters:: component (str) – Channel component name to retrieve (e.g., ‘ex’, ‘hy’).
Returns:: Channel object if found, None otherwise. Return type depends on the channel type.
Return type:: Electric | Magnetic | Auxiliary | None

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

See also

has_channel: Check if channel exists
add_channel: Add a channel to the run

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.

Parameters:

channel_obj (Electric | Magnetic | Auxiliary | dict | str) –
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
update (bool, optional) – 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).

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)

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

See also

remove_channel: Remove a channel from the run
get_channel: Retrieve a channel object
update_time_period: Manually update time period

remove_channel(channel_id)

Remove a channel from the run.

Parameters:: channel_id (str) – Channel component name to remove (e.g., ‘ex’, ‘hy’).

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.

Notes

Automatically updates the channels_recorded lists after removal. Logs a warning if the channel is not found.

See also

add_channel: Add a channel to the run
has_channel: Check if channel exists

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.
Return type:: dict[str, str]

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'}

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.

See also

add_channel: Add channels to the run
get_channel: Access channels by component

property n_channels: int

Number of channels in the run.

Returns:: Count of channels currently in the run.
Return type:: int

Examples

>>> run = Run(id='001')
>>> print(run.n_channels)
0
>>> run.add_channel('ex')
>>> run.add_channel('hy')
>>> print(run.n_channels)
2

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’.

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

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

See also

add_channel: Add channel and optionally update time period