Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ run.py: 81%
215 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# =====================================================
2# Imports
3# =====================================================
4from __future__ import annotations
6from collections import OrderedDict
7from typing import Annotated
9import numpy as np
10from loguru import logger
11from pydantic import (
12 computed_field,
13 Field,
14 field_validator,
15 model_validator,
16 ValidationInfo,
17)
18from typing_extensions import Self
20from mt_metadata.base import MetadataBase
21from mt_metadata.common import (
22 AuthorPerson,
23 Comment,
24 DataTypeEnum,
25 Fdsn,
26 Provenance,
27 TimePeriod,
28)
29from mt_metadata.common.list_dict import ListDict
30from mt_metadata.timeseries import Auxiliary, DataLogger, Electric, Magnetic
33# =====================================================
36class Run(MetadataBase):
37 channels_recorded_auxiliary: Annotated[
38 list[str],
39 Field(
40 default_factory=list,
41 description="List of auxiliary channels recorded",
42 alias=None,
43 json_schema_extra={
44 "units": None,
45 "required": True,
46 "examples": ["[T]"],
47 },
48 ),
49 ]
51 channels_recorded_electric: Annotated[
52 list[str],
53 Field(
54 default_factory=list,
55 description="List of electric channels recorded",
56 alias=None,
57 json_schema_extra={
58 "units": None,
59 "required": True,
60 "examples": ["[Ex , Ey]"],
61 },
62 ),
63 ]
65 channels_recorded_magnetic: Annotated[
66 list[str],
67 Field(
68 default_factory=list,
69 description="List of magnetic channels recorded",
70 alias=None,
71 json_schema_extra={
72 "units": None,
73 "required": True,
74 "examples": ["[Hx , Hy , Hz]"],
75 },
76 ),
77 ]
79 @computed_field
80 def channels_recorded_all(self) -> list[str]:
81 """
82 List of all channels recorded in the run.
83 """
84 return sorted(
85 [ch.component for ch in self.channels.values() if ch.component is not None]
86 )
88 comments: Annotated[
89 Comment,
90 Field(
91 default_factory=Comment, # type: ignore
92 description="Any comments on the run.",
93 alias=None,
94 json_schema_extra={
95 "units": None,
96 "required": False,
97 "examples": ["cows chewed cables"],
98 },
99 ),
100 ]
102 data_type: Annotated[
103 DataTypeEnum,
104 Field(
105 default=DataTypeEnum.BBMT,
106 description="Type of data recorded for this run.",
107 alias=None,
108 json_schema_extra={
109 "units": None,
110 "required": True,
111 "examples": ["BBMT"],
112 },
113 ),
114 ]
116 id: Annotated[
117 str,
118 Field(
119 default="",
120 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.",
121 alias=None,
122 pattern="^[a-zA-Z0-9_]*$",
123 json_schema_extra={
124 "units": None,
125 "required": True,
126 "examples": ["001"],
127 },
128 ),
129 ]
131 sample_rate: Annotated[
132 float,
133 Field(
134 default=0.0,
135 description="Digital sample rate for the run",
136 alias=None,
137 json_schema_extra={
138 "units": "samples per second",
139 "required": True,
140 "examples": ["100"],
141 },
142 ),
143 ]
145 acquired_by: Annotated[
146 AuthorPerson,
147 Field(
148 default_factory=AuthorPerson, # type: ignore
149 description="Information about the group that collected the data.",
150 alias=None,
151 json_schema_extra={
152 "units": None,
153 "required": False,
154 "examples": ["Person()"],
155 },
156 ),
157 ]
159 metadata_by: Annotated[
160 AuthorPerson,
161 Field(
162 default_factory=AuthorPerson, # type: ignore
163 description="Information about the group that collected the metadata.",
164 alias=None,
165 json_schema_extra={
166 "units": None,
167 "required": False,
168 "examples": ["Person()"],
169 },
170 ),
171 ]
173 provenance: Annotated[
174 Provenance,
175 Field(
176 default_factory=Provenance, # type: ignore
177 description="Provenance information about the run.",
178 alias=None,
179 json_schema_extra={
180 "units": None,
181 "required": False,
182 "examples": ["Provenance()"],
183 },
184 ),
185 ]
187 time_period: Annotated[
188 TimePeriod,
189 Field(
190 default_factory=TimePeriod, # type: ignore
191 description="Time period for the run.",
192 alias=None,
193 json_schema_extra={
194 "units": None,
195 "required": False,
196 "examples": ["TimePeriod(start='2020-01-01', end='2020-12-31')"],
197 },
198 ),
199 ]
201 data_logger: Annotated[
202 DataLogger,
203 Field(
204 default_factory=DataLogger, # type: ignore
205 description="Data Logger information used to collect the run.",
206 alias=None,
207 json_schema_extra={
208 "units": None,
209 "required": False,
210 "examples": ["DataLogger()"],
211 },
212 ),
213 ]
215 fdsn: Annotated[
216 Fdsn,
217 Field(
218 default_factory=Fdsn, # type: ignore
219 description="FDSN information for the run.",
220 alias=None,
221 json_schema_extra={
222 "units": None,
223 "required": False,
224 "examples": ["Fdsn()"],
225 },
226 ),
227 ]
229 channels: Annotated[
230 ListDict | list | dict | OrderedDict,
231 Field(
232 default_factory=ListDict,
233 description="ListDict of channel objects collected in this run.",
234 alias=None,
235 exclude=True,
236 json_schema_extra={
237 "units": None,
238 "required": False,
239 "examples": ["ListDict(Electric(), Magnetic(), Auxiliary())"],
240 },
241 ),
242 ]
244 @field_validator("comments", mode="before")
245 @classmethod
246 def validate_comments(cls, value, info: ValidationInfo) -> Comment:
247 """
248 Validate that the value is a valid comment.
249 """
250 if isinstance(value, str):
251 return Comment(value=value)
252 return value
254 @field_validator("data_type", mode="before")
255 @classmethod
256 def validate_data_type(cls, value, info: ValidationInfo) -> str:
257 """
258 Validate that the data_type is a string.
259 """
260 if isinstance(value, DataTypeEnum):
261 value = value.value
262 elif isinstance(value, str):
263 if "," in value:
264 value = value.split(",")[0].strip()
265 elif not isinstance(value, str):
266 raise TypeError(f"data_type must be a string, not {type(value)}")
267 return value
269 @field_validator(
270 "channels_recorded_electric",
271 "channels_recorded_magnetic",
272 "channels_recorded_auxiliary",
273 mode="before",
274 )
275 @classmethod
276 def validate_list_of_strings(
277 cls, value: np.ndarray | list[str] | str, info: ValidationInfo
278 ) -> list[str]:
279 """
280 Validate that the value is a list of strings.
281 """
282 if value in [None, "None", "none", "NONE", "null"]:
283 return []
285 if isinstance(value, np.ndarray):
286 value = value.astype(str).tolist()
288 elif isinstance(value, (list, tuple)):
289 value = [str(v) for v in value]
291 elif isinstance(value, (str)):
292 value = [v.strip() for v in value.split(",")]
294 else:
295 raise TypeError(
296 "'channels_recorded' must be set with a list of strings not "
297 f"{type(value)}."
298 )
299 return value
301 @model_validator(mode="after")
302 def validate_channels_recorded(self) -> Self:
303 """
304 Validate that the value is a list of strings.
305 """
306 # need to make each another object list() otherwise the contents
307 # get overwritten with the new channel.
308 for electric in list(self.channels_recorded_electric):
309 if electric not in self.channels.keys():
310 self.add_channel(Electric(component=electric)) # type: ignore
311 for magnetic in list(self.channels_recorded_magnetic):
312 if magnetic not in self.channels.keys():
313 self.add_channel(Magnetic(component=magnetic)) # type: ignore
314 for auxiliary in list(self.channels_recorded_auxiliary):
315 if auxiliary not in self.channels.keys():
316 self.add_channel(Auxiliary(component=auxiliary)) # type: ignore
317 return self
319 @field_validator("channels", mode="before")
320 @classmethod
321 def validate_channels(cls, value, info: ValidationInfo) -> ListDict:
322 if not isinstance(value, (list, tuple, dict, ListDict, OrderedDict)):
323 msg = (
324 "input run_list must be an iterable, should be a list or dict "
325 f"not {type(value)}"
326 )
327 logger.error(msg)
328 raise TypeError(msg)
330 fails = []
331 channels = ListDict()
332 if isinstance(value, (dict, ListDict, OrderedDict)):
333 value_list = value.values()
335 elif isinstance(value, (list, tuple)):
336 value_list = value
338 for ii, channel in enumerate(value_list):
339 try:
340 channels.append(cls._get_correct_channel_type(channel))
341 except KeyError as error:
342 msg = f"Could not find type in {channel}"
343 logger.error(msg)
344 logger.exception(error)
345 fails.append(msg)
347 if len(fails) > 0:
348 raise TypeError("\n".join(fails))
350 return channels
352 def _empty_channels_recorded(self) -> None:
353 """
354 Clear all channels recorded lists.
356 Removes all entries from the auxiliary, electric, and magnetic
357 channels recorded lists. This is typically called before updating
358 the lists based on current channel contents.
360 See Also
361 --------
362 _update_channels_recorded : Update channels recorded lists from current channels
364 """
365 self.channels_recorded_auxiliary.clear()
366 self.channels_recorded_electric.clear()
367 self.channels_recorded_magnetic.clear()
369 def _update_channels_recorded(self) -> None:
370 """
371 Update channels recorded lists based on current channels.
373 Scans all channels in the run and populates the appropriate
374 channels_recorded lists (auxiliary, electric, magnetic) based on
375 channel types and components. Excludes default/None components.
377 Notes
378 -----
379 This method is automatically called when channels are added or removed.
380 The lists are sorted alphabetically.
382 Excluded components:
384 - Auxiliary: None, 'auxiliary_default'
385 - Electric: None, 'e_default'
386 - Magnetic: None, 'h_default'
388 Examples
389 --------
390 >>> run = Run(id='001')
391 >>> run.add_channel(Electric(component='ex'))
392 >>> run.add_channel(Magnetic(component='hx'))
393 >>> # _update_channels_recorded is called automatically
394 >>> print(run.channels_recorded_electric)
395 ['ex']
396 >>> print(run.channels_recorded_magnetic)
397 ['hx']
399 """
400 self._empty_channels_recorded()
401 aux_components = [
402 ch.component
403 for ch in self.channels
404 if isinstance(ch, Auxiliary)
405 and ch.component not in [None, "auxiliary_default"]
406 ]
407 elec_components = [
408 ch.component
409 for ch in self.channels
410 if isinstance(ch, Electric) and ch.component not in [None, "e_default"]
411 ]
412 mag_components = [
413 ch.component
414 for ch in self.channels
415 if isinstance(ch, Magnetic) and ch.component not in [None, "h_default"]
416 ]
417 self.channels_recorded_auxiliary = sorted(aux_components)
418 self.channels_recorded_electric = sorted(elec_components)
419 self.channels_recorded_magnetic = sorted(mag_components)
421 # def __len__(self):
422 # return len(self.channels)
424 def merge(self, other: Run, inplace: bool = True) -> Run | None:
425 """
426 Merge channels from another Run into this run.
428 Combines channels from two runs and updates the channels recorded lists
429 and time period.
431 Parameters
432 ----------
433 other : Run
434 Another Run object whose channels will be merged into this run.
435 inplace : bool, optional
436 If True, update this run and update time period. If False, return
437 a copy of the merged run (default is True).
439 Returns
440 -------
441 Run | None
442 If inplace is False, returns a copy of the merged Run. Otherwise None.
444 Raises
445 ------
446 TypeError
447 If other is not a Run object.
449 Examples
450 --------
451 Merge runs in place:
453 >>> run1 = Run(id='001')
454 >>> run1.add_channel(Electric(component='ex'))
455 >>> run2 = Run(id='002')
456 >>> run2.add_channel(Magnetic(component='hx'))
457 >>> run1.merge(run2, inplace=True)
458 >>> print(run1.channels_recorded_all)
459 ['ex', 'hx']
461 Merge and return new run:
463 >>> merged_run = run1.merge(run2, inplace=False)
464 >>> print(merged_run.channels_recorded_all)
465 ['ex', 'hx']
467 See Also
468 --------
469 update : Update metadata from another run
471 """
472 if isinstance(other, Run):
473 self.channels.extend(other.channels)
474 self._update_channels_recorded()
475 if inplace:
476 self.update_time_period()
477 else:
478 return self.copy()
479 else:
480 msg = f"Can only merge Run objects, not {type(other)}"
481 logger.error(msg)
482 raise TypeError(msg)
484 def update(self, other: Run, match: list[str] | None = []) -> None:
485 """
486 Update attribute values from another Run object.
488 Copies non-None, non-default attribute values from another Run object
489 to this one. Skips empty values like None, 0.0, [], empty strings,
490 and default timestamps.
492 Parameters
493 ----------
494 other : Run
495 Another Run object to copy attributes from.
496 match : list[str] | None, optional
497 List of attribute names that must match between runs before updating.
498 If any don't match, raises ValueError. Typically used for 'id' to
499 ensure runs are compatible (default is None).
501 Raises
502 ------
503 ValueError
504 If any attributes in match list don't have equal values.
505 TypeError
506 If other is not a compatible Run type.
508 Examples
509 --------
510 Basic update:
512 >>> run1 = Run(id='001', sample_rate=256.0)
513 >>> run2 = Run(id='001', sample_rate=0.0)
514 >>> run2.acquired_by.author = 'J. Doe'
515 >>> run1.update(run2)
516 >>> print(run1.acquired_by.author)
517 'J. Doe'
518 >>> print(run1.sample_rate) # Not updated (run2 has default 0.0)
519 256.0
521 Update with matching check:
523 >>> run1 = Run(id='001')
524 >>> run2 = Run(id='002')
525 >>> try:
526 ... run1.update(run2, match=['id'])
527 ... except ValueError as e:
528 ... print("IDs don't match!")
529 IDs don't match!
531 Notes
532 -----
533 Channel metadata is also updated. For each channel in other, if the
534 channel exists in this run, it's updated; if not, it's added.
536 Skipped values:
538 - None
539 - 0.0
540 - Empty lists []
541 - Empty strings ''
542 - Default timestamp '1980-01-01T00:00:00+00:00'
544 See Also
545 --------
546 merge : Merge channels from another run
548 """
549 # Check if other is a compatible Run type (handles dynamically created classes)
550 if not (
551 isinstance(other, type(self))
552 or (hasattr(other, "__class__") and other.__class__.__name__ == "Run")
553 ):
554 logger.warning(f"Cannot update {type(self)} with {type(other)}")
555 for k in match:
556 if self.get_attr_from_name(k) != other.get_attr_from_name(k):
557 msg = (
558 f"{k} is not equal {self.get_attr_from_name(k)} != "
559 "{other.get_attr_from_name(k)}"
560 )
561 logger.error(msg)
562 raise ValueError(msg)
563 for k, v in other.to_dict(single=True).items():
564 if hasattr(v, "size"):
565 if v.size > 0:
566 self.update_attribute(k, v)
567 else:
568 if v not in [None, 0.0, [], "", "1980-01-01T00:00:00+00:00"]:
569 self.update_attribute(k, v)
571 ## Need this because channels are set when setting channels_recorded
572 ## and it initiates an empty channel, but we need to fill it with
573 ## the appropriate metadata.
574 for ch in other.channels:
575 self.add_channel(ch)
577 def has_channel(self, component: str) -> bool:
578 """
579 Check if a channel with the given component exists in the run.
581 Parameters
582 ----------
583 component : str
584 Channel component name to search for (e.g., 'ex', 'hy').
586 Returns
587 -------
588 bool
589 True if channel exists, False otherwise.
591 Examples
592 --------
593 >>> run = Run(id='001')
594 >>> run.add_channel(Electric(component='ex'))
595 >>> print(run.has_channel('ex'))
596 True
597 >>> print(run.has_channel('ey'))
598 False
600 See Also
601 --------
602 get_channel : Retrieve a channel object
603 channel_index : Get the index of a channel
605 """
607 if component in [cc for cc in self.channels_recorded_all]:
608 return True
609 return False
611 def channel_index(self, component: str) -> int | None:
612 """
613 Get the index of a channel in the channels_recorded_all list.
615 Parameters
616 ----------
617 component : str
618 Channel component name to search for (e.g., 'ex', 'hy').
620 Returns
621 -------
622 int | None
623 Index of the channel if found, None otherwise.
625 Examples
626 --------
627 >>> run = Run(id='001')
628 >>> run.add_channel(Electric(component='ex'))
629 >>> run.add_channel(Electric(component='ey'))
630 >>> run.add_channel(Magnetic(component='hx'))
631 >>> print(run.channel_index('ey'))
632 1
633 >>> print(run.channel_index('hz'))
634 None
636 Notes
637 -----
638 Channels are sorted alphabetically in channels_recorded_all.
640 See Also
641 --------
642 has_channel : Check if channel exists
643 get_channel : Retrieve channel object
645 """
646 if self.has_channel(component):
647 return self.channels_recorded_all.index(component)
648 return None
650 def get_channel(self, component: str) -> Electric | Magnetic | Auxiliary | None:
651 """
652 Retrieve a channel object by component name.
654 Parameters
655 ----------
656 component : str
657 Channel component name to retrieve (e.g., 'ex', 'hy').
659 Returns
660 -------
661 Electric | Magnetic | Auxiliary | None
662 Channel object if found, None otherwise. Return type depends on
663 the channel type.
665 Examples
666 --------
667 >>> run = Run(id='001')
668 >>> ex = Electric(component='ex', dipole_length=100.0)
669 >>> run.add_channel(ex)
670 >>> channel = run.get_channel('ex')
671 >>> print(type(channel).__name__)
672 'Electric'
673 >>> print(channel.dipole_length)
674 100.0
675 >>> print(run.get_channel('ey'))
676 None
678 See Also
679 --------
680 has_channel : Check if channel exists
681 add_channel : Add a channel to the run
683 """
685 if self.has_channel(component):
686 return self.channels[component]
688 def add_channel(
689 self,
690 channel_obj: Electric | Magnetic | Auxiliary | dict | str,
691 update: bool = True,
692 ) -> None:
693 """
694 Add or update a channel in the run.
696 If the channel already exists (matched by component), its metadata
697 is updated. If it doesn't exist, it's added to the channels list.
698 Can accept channel objects, dictionaries, or component strings.
700 Parameters
701 ----------
702 channel_obj : Electric | Magnetic | Auxiliary | dict | str
703 Channel to add. Can be:
705 - Channel object (Electric, Magnetic, or Auxiliary)
706 - Dictionary with channel attributes (must include 'type' or 'component')
707 - String component name (e.g., 'ex', 'hy', 'temp')
709 If string, channel type is inferred:
711 - Starts with 'e' → Electric
712 - Starts with 'h' or 'b' or equals 'magnetic' → Magnetic
713 - Otherwise → Auxiliary
715 update : bool, optional
716 If True, update the run's time period to include this channel's
717 time period. If False, don't update time period (default is True).
719 Examples
720 --------
721 Add channel objects:
723 >>> run = Run(id='001')
724 >>> ex = Electric(component='ex', dipole_length=100.0)
725 >>> run.add_channel(ex)
726 >>> print(run.channels_recorded_electric)
727 ['ex']
729 Add from string (infers type):
731 >>> run.add_channel('hy')
732 >>> run.add_channel('temperature')
733 >>> print(run.channels_recorded_magnetic)
734 ['hy']
735 >>> print(run.channels_recorded_auxiliary)
736 ['temperature']
738 Add from dictionary:
740 >>> channel_dict = {
741 ... 'type': 'electric',
742 ... 'component': 'ey',
743 ... 'dipole_length': 95.0
744 ... }
745 >>> run.add_channel(channel_dict)
747 Update existing channel:
749 >>> ex_updated = Electric(component='ex', dipole_length=105.0)
750 >>> run.add_channel(ex_updated) # Updates existing 'ex'
751 >>> print(run.get_channel('ex').dipole_length)
752 105.0
754 Add without updating time period:
756 >>> run.add_channel('hz', update=False)
758 Notes
759 -----
760 This method automatically:
762 - Updates channels_recorded lists
763 - Updates run time period (if update=True)
764 - Converts string/dict inputs to proper channel objects
765 - Logs when updating existing channels
767 See Also
768 --------
769 remove_channel : Remove a channel from the run
770 get_channel : Retrieve a channel object
771 update_time_period : Manually update time period
773 """
774 channel_obj = self._get_correct_channel_type(channel_obj)
776 if self.has_channel(channel_obj.component):
777 self.channels[channel_obj.component].update(channel_obj)
778 logger.debug(
779 f"Run {channel_obj.component} already exists, updating metadata"
780 )
782 else:
783 self.channels.append(channel_obj)
785 self._update_channels_recorded()
787 if update:
788 self.update_time_period()
790 def remove_channel(self, channel_id: str) -> None:
791 """
792 Remove a channel from the run.
794 Parameters
795 ----------
796 channel_id : str
797 Channel component name to remove (e.g., 'ex', 'hy').
799 Examples
800 --------
801 >>> run = Run(id='001')
802 >>> run.add_channel(Electric(component='ex'))
803 >>> run.add_channel(Electric(component='ey'))
804 >>> print(run.channels_recorded_electric)
805 ['ex', 'ey']
806 >>> run.remove_channel('ex')
807 >>> print(run.channels_recorded_electric)
808 ['ey']
809 >>> run.remove_channel('ez') # Doesn't exist
810 # Logs warning: Could not find ez to remove.
812 Notes
813 -----
814 Automatically updates the channels_recorded lists after removal.
815 Logs a warning if the channel is not found.
817 See Also
818 --------
819 add_channel : Add a channel to the run
820 has_channel : Check if channel exists
822 """
824 if self.has_channel(channel_id):
825 self.channels.remove(channel_id)
827 self._update_channels_recorded()
828 else:
829 logger.warning(f"Could not find {channel_id} to remove.")
831 def update_channel_keys(self) -> dict[str, str]:
832 """
833 Update channel dictionary keys to match current component values.
835 Updates the keys in the channels ListDict to match current channel
836 components. Useful when channel components have been modified after
837 channels were added, ensuring channels can be accessed by their
838 current component values.
840 Returns
841 -------
842 dict[str, str]
843 Mapping of old keys to new keys showing what was changed.
845 Examples
846 --------
847 Fix keys after modifying components:
849 >>> run = Run(id='001')
850 >>> channel = Electric(component='')
851 >>> run.add_channel(channel)
852 >>> # Channel is stored with empty string key
853 >>> channel.component = 'ex'
854 >>> key_mapping = run.update_channel_keys()
855 >>> print(key_mapping)
856 {'': 'ex'}
857 >>> # Now accessible as run.channels['ex']
858 >>> print(run.get_channel('ex').component)
859 'ex'
861 Multiple key updates:
863 >>> run = Run(id='001')
864 >>> ch1 = Electric(component='e1')
865 >>> ch2 = Magnetic(component='h1')
866 >>> run.add_channel(ch1)
867 >>> run.add_channel(ch2)
868 >>> ch1.component = 'ex'
869 >>> ch2.component = 'hx'
870 >>> mapping = run.update_channel_keys()
871 >>> print(mapping)
872 {'e1': 'ex', 'h1': 'hx'}
874 Notes
875 -----
876 This is typically only needed if you've directly modified channel
877 component attributes after adding them to the run. Normal usage
878 doesn't require calling this method.
880 See Also
881 --------
882 add_channel : Add channels to the run
883 get_channel : Access channels by component
885 """
886 return self.channels.update_keys()
888 @property
889 def n_channels(self) -> int:
890 """
891 Number of channels in the run.
893 Returns
894 -------
895 int
896 Count of channels currently in the run.
898 Examples
899 --------
900 >>> run = Run(id='001')
901 >>> print(run.n_channels)
902 0
903 >>> run.add_channel('ex')
904 >>> run.add_channel('hy')
905 >>> print(run.n_channels)
906 2
907 """
908 return len(self.channels)
910 def update_time_period(self) -> None:
911 """
912 Update run's time period to encompass all channel time periods.
914 Examines all channels in the run and updates the run's start and end
915 times to include the earliest start and latest end from all channels.
916 Ignores default timestamp '1980-01-01T00:00:00+00:00'.
918 Examples
919 --------
920 >>> from mt_metadata.timeseries import Run, Electric
921 >>> run = Run(id='001')
922 >>> ex = Electric(component='ex')
923 >>> ex.time_period.start = '2020-01-01T00:00:00+00:00'
924 >>> ex.time_period.end = '2020-01-01T01:00:00+00:00'
925 >>> run.add_channel(ex, update=False)
926 >>> print(run.time_period.start)
927 1980-01-01T00:00:00+00:00
928 >>> run.update_time_period()
929 >>> print(run.time_period.start)
930 2020-01-01T00:00:00+00:00
932 Multiple channels:
934 >>> ey = Electric(component='ey')
935 >>> ey.time_period.start = '2020-01-01T00:30:00+00:00'
936 >>> ey.time_period.end = '2020-01-01T02:00:00+00:00'
937 >>> run.add_channel(ey, update=True)
938 >>> print(run.time_period.start) # Uses earliest
939 2020-01-01T00:00:00+00:00
940 >>> print(run.time_period.end) # Uses latest
941 2020-01-01T02:00:00+00:00
943 Notes
944 -----
945 - Only updates if channels exist (n_channels > 0)
946 - Ignores channels with default timestamp
947 - Always expands time period, never shrinks it
948 - Automatically called by add_channel() when update=True
950 See Also
951 --------
952 add_channel : Add channel and optionally update time period
954 """
955 if self.n_channels > 0:
956 start = []
957 end = []
958 for channel in self.channels:
959 if channel.time_period.start != "1980-01-01T00:00:00+00:00":
960 start.append(channel.time_period.start)
961 if channel.time_period.end != "1980-01-01T00:00:00+00:00":
962 end.append(channel.time_period.end)
964 if start:
965 if self.time_period.start == "1980-01-01T00:00:00+00:00":
966 self.time_period.start = min(start)
967 else:
968 if self.time_period.start > min(start):
969 self.time_period.start = min(start)
971 if end:
972 if self.time_period.end == "1980-01-01T00:00:00+00:00":
973 self.time_period.end = max(end)
974 else:
975 if self.time_period.end < max(end):
976 self.time_period.end = max(end)
978 @classmethod
979 def _get_correct_channel_type(
980 cls, channel_obj: Electric | Magnetic | Auxiliary | dict | OrderedDict | str
981 ) -> Electric | Magnetic | Auxiliary:
982 """
983 Convert input to the correct channel type object.
985 Determines the appropriate channel type (Electric, Magnetic, or
986 Auxiliary) based on the input and returns a properly typed channel
987 object. Handles string, dict, and object inputs.
989 Parameters
990 ----------
991 channel_obj : Electric | Magnetic | Auxiliary | dict | OrderedDict | str
992 Input to convert to a channel object. Can be:
994 - Channel object: returned as-is (after validation)
995 - String: component name, type inferred from first letter
996 - Dict: must contain 'type' or 'component' key
998 Returns
999 -------
1000 Electric | Magnetic | Auxiliary
1001 Properly typed channel object.
1003 Raises
1004 ------
1005 ValueError
1006 If channel_obj is a channel with None component (except Auxiliary),
1007 or if input type is not supported.
1008 KeyError
1009 If dict input doesn't contain required keys.
1011 Examples
1012 --------
1013 From string:
1015 >>> channel = Run._get_correct_channel_type('ex')
1016 >>> print(type(channel).__name__)
1017 'Electric'
1018 >>> print(channel.component)
1019 'ex'
1021 From dict:
1023 >>> ch_dict = {'type': 'magnetic', 'component': 'hx'}
1024 >>> channel = Run._get_correct_channel_type(ch_dict)
1025 >>> print(type(channel).__name__)
1026 'Magnetic'
1028 Type inference from component letter:
1030 >>> Run._get_correct_channel_type('ey') # 'e' → Electric
1031 <Electric ...>
1032 >>> Run._get_correct_channel_type('hz') # 'h' → Magnetic
1033 <Magnetic ...>
1034 >>> Run._get_correct_channel_type('temp') # other → Auxiliary
1035 <Auxiliary ...>
1037 Notes
1038 -----
1039 Type inference rules:
1041 **From string component:**
1043 - Starts with 'e' → Electric
1044 - Starts with 'h', 'b', or equals 'magnetic' → Magnetic
1045 - Anything else → Auxiliary
1047 **From dict:**
1049 - Uses 'type' key if present, otherwise first letter of 'component'
1050 - Same inference rules apply
1052 See Also
1053 --------
1054 add_channel : Add a channel using this type detection
1056 """
1057 if isinstance(channel_obj, str):
1058 if channel_obj.lower().startswith("e"):
1059 return Electric(component=channel_obj)
1060 elif (
1061 channel_obj.lower().startswith("h")
1062 or channel_obj.lower().startswith("b")
1063 or channel_obj.lower() in ["magnetic"]
1064 ):
1065 return Magnetic(component=channel_obj)
1066 else:
1067 return Auxiliary(component=channel_obj)
1069 elif isinstance(channel_obj, (dict, OrderedDict)):
1070 try:
1071 ch_type = channel_obj["type"]
1072 if ch_type is None:
1073 ch_type = channel_obj["component"][0]
1075 if ch_type.lower().startswith("e"):
1076 return Electric(**channel_obj)
1077 elif (
1078 ch_type.lower().startswith("b")
1079 or ch_type.lower().startswith("b")
1080 or ch_type.lower() in ["magnetic"]
1081 ):
1082 return Magnetic(**channel_obj)
1083 else:
1084 return Auxiliary(**channel_obj)
1085 except KeyError as error:
1086 msg = f"Could not find type in {channel_obj}"
1087 logger.error(msg)
1088 logger.exception(error)
1089 raise error
1091 elif isinstance(channel_obj, (Electric, Magnetic, Auxiliary)):
1092 if channel_obj.component is None:
1093 if not isinstance(channel_obj, Auxiliary):
1094 msg = "component cannot be empty"
1095 logger.error(msg)
1096 raise ValueError(msg)
1097 return channel_obj
1098 else:
1099 msg = (
1100 f"Input must be mt_metadata.timeseries.Channel not {type(channel_obj)}"
1101 )
1102 logger.error(msg)
1103 raise ValueError(msg)