Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ transfer_functions \ io \ emtfxml \ metadata \ run.py: 92%
77 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# =====================================================
2# Imports
3# =====================================================
4from typing import Annotated
6import numpy as np
7import pandas as pd
8from loguru import logger
9from pydantic import Field, field_validator
11from mt_metadata.base import MetadataBase
12from mt_metadata.base.helpers import element_to_string
13from mt_metadata.common import Comment
14from mt_metadata.common.mttime import MTime
15from mt_metadata.transfer_functions.io.emtfxml.metadata import helpers
17from . import Dipole, Instrument, Magnetometer
20# =====================================================
21class Run(MetadataBase):
22 errors: Annotated[
23 str | None,
24 Field(
25 default=None,
26 description="Any field errors",
27 alias=None,
28 json_schema_extra={
29 "units": None,
30 "required": False,
31 "examples": ["moose ate cables"],
32 },
33 ),
34 ]
36 run: Annotated[
37 str,
38 Field(
39 default="",
40 description="Run name",
41 alias=None,
42 json_schema_extra={
43 "units": None,
44 "required": True,
45 "examples": ["mt001a"],
46 },
47 ),
48 ]
50 sampling_rate: Annotated[
51 float | None,
52 Field(
53 default=None,
54 description="Sample rate of the run",
55 alias=None,
56 json_schema_extra={
57 "units": "samples per second",
58 "required": False,
59 "examples": ["1"],
60 },
61 ),
62 ]
64 start: Annotated[
65 MTime | str | float | int | np.datetime64 | pd.Timestamp,
66 Field(
67 default_factory=lambda: MTime(time_stamp=None),
68 description="Date time when the data collection started",
69 alias=None,
70 json_schema_extra={
71 "units": None,
72 "required": True,
73 "examples": ["2020-01-01T12:00:00"],
74 },
75 ),
76 ]
78 end: Annotated[
79 MTime | str | float | int | np.datetime64 | pd.Timestamp,
80 Field(
81 default_factory=lambda: MTime(time_stamp=None),
82 description="Date time when the data collection ended",
83 alias=None,
84 json_schema_extra={
85 "units": None,
86 "required": True,
87 "examples": ["2020-05-01T12:00:00"],
88 },
89 ),
90 ]
91 comments: Annotated[
92 Comment | str | None,
93 Field(
94 default_factory=Comment, # type: ignore
95 description="Comments about the run",
96 alias=None,
97 json_schema_extra={
98 "units": None,
99 "required": False,
100 "examples": ["Comment(text='This is a comment')"],
101 },
102 ),
103 ]
105 instrument: Annotated[
106 Instrument,
107 Field(
108 default_factory=Instrument, # type: ignore
109 description="Instrument used for the run",
110 alias=None,
111 json_schema_extra={
112 "units": None,
113 "required": False,
114 "examples": ["Instrument(name='MT Sensor', type='magnetometer')"],
115 },
116 ),
117 ]
119 magnetometer: Annotated[
120 list[Magnetometer],
121 Field(
122 default_factory=list,
123 description="List of magnetometers used in the run",
124 alias=None,
125 json_schema_extra={
126 "units": None,
127 "required": False,
128 "examples": ["Magnetometer(name='Magnetometer 1', type='fluxgate')"],
129 },
130 ),
131 ]
132 dipole: Annotated[
133 list[Dipole],
134 Field(
135 default_factory=list,
136 description="List of dipoles used in the run",
137 alias=None,
138 json_schema_extra={
139 "units": None,
140 "required": False,
141 "examples": ["Dipole(name='Dipole 1', type='fluxgate')"],
142 },
143 ),
144 ]
146 @field_validator("start", "end", mode="before")
147 @classmethod
148 def validate_start(
149 cls, field_value: MTime | float | int | np.datetime64 | pd.Timestamp | str
150 ):
151 return MTime(time_stamp=field_value)
153 @field_validator("comments", mode="before")
154 @classmethod
155 def validate_comments(cls, field_value: Comment | str | None):
156 if isinstance(field_value, str):
157 return Comment(value=field_value)
158 return field_value
160 def read_dict(self, input_dict: dict) -> None:
161 """
162 Field notes are odd so have a special reader to do it piece by
163 painstaking piece.
165 :param input_dict: input dictionary containing run data
166 :type input_dict: dict
167 :return: None
168 :rtype: None
170 """
172 self.run = input_dict["run"]
173 self.instrument.from_dict({"instrument": input_dict["instrument"]})
174 self.sampling_rate = input_dict["sampling_rate"]
175 self.start = input_dict["start"]
176 self.end = input_dict["end"]
177 try:
178 if isinstance(input_dict["comments"], list):
179 self.comments.from_dict({"comments": input_dict["comments"][0]})
180 else:
181 self.comments.from_dict({"comments": input_dict["comments"]})
182 except KeyError:
183 logger.debug("run has no comments")
184 self.errors = input_dict["errors"]
186 try:
187 if isinstance(input_dict["magnetometer"], list):
188 self.magnetometer = []
189 for mag in input_dict["magnetometer"]:
190 m = Magnetometer()
191 m.from_dict({"magnetometer": mag})
192 self.magnetometer.append(m)
193 else:
194 self.magnetometer = []
195 m = Magnetometer()
196 m.from_dict({"magnetometer": input_dict["magnetometer"]})
197 self.magnetometer.append(m)
198 except KeyError:
199 logger.debug("run has no magnetotmeter information")
201 try:
202 if isinstance(input_dict["dipole"], list):
203 self.dipole = []
204 for mag in input_dict["dipole"]:
205 m = Dipole()
206 m.from_dict({"dipole": mag})
207 self.dipole.append(m)
208 else:
209 m = Dipole()
210 m.from_dict({"dipole": input_dict["dipole"]})
211 self.dipole.append(m)
212 except KeyError:
213 logger.debug("run has no dipole information")
215 def to_xml(self, string=False, required=True):
216 """
218 :param string: DESCRIPTION, defaults to True
219 :type string: TYPE, optional
220 :param required: DESCRIPTION, defaults to False
221 :type required: TYPE, optional
222 :return: DESCRIPTION
223 :rtype: TYPE
225 """
226 element = helpers.to_xml(
227 self,
228 string=False,
229 required=required,
230 order=[
231 "instrument",
232 "magnetometer",
233 "dipole",
234 "comments",
235 "errors",
236 "sampling_rate",
237 "start",
238 "end",
239 ],
240 )
241 element.attrib = {"run": self.run}
242 element.tag = "field_notes"
244 element.find("SamplingRate").attrib["units"] = "Hz"
246 if string:
247 return element_to_string(element)
248 return element