Skip to content

Commit 611aa64

Browse files
felipenarvfelipenarv
committed
Added mapper and test for it
1 parent cca387f commit 611aa64

2 files changed

Lines changed: 429 additions & 0 deletions

File tree

‎src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py‎

Lines changed: 284 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,284 @@
1+
from dataclasses import dataclass
2+
3+
from allotropy.allotrope.converter import add_custom_information_document
4+
from allotropy.allotrope.models.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import (
5+
DataProcessingAggregateDocument,
6+
DataProcessingDocumentItem,
7+
DetectorControlAggregateDocument,
8+
DetectorControlDocumentItem,
9+
DeviceDocumentItem,
10+
DeviceSystemDocument,
11+
MassSpectrometryAggregateDocument,
12+
MassSpectrometryDocumentItem,
13+
MeasurementDocument,
14+
Model,
15+
PeakItem,
16+
PeakList,
17+
ProcessedDataAggregateDocument,
18+
ProcessedDataDocumentItem,
19+
SampleDocument,
20+
SampleIntroductionDocument,
21+
)
22+
from allotropy.allotrope.models.shared.definitions.custom import (
23+
TQuantityValueDalton,
24+
TQuantityValueHertz,
25+
TQuantityValueMassPerCharge,
26+
TQuantityValueMilliliterPerMinute,
27+
TQuantityValuePercent,
28+
TQuantityValueSecondTime,
29+
)
30+
from allotropy.allotrope.models.shared.definitions.definitions import TQuantityValue
31+
from allotropy.allotrope.schema_mappers.schema_mapper import SchemaMapper
32+
from allotropy.parsers.utils.values import quantity_or_none
33+
34+
35+
@dataclass(frozen=True)
36+
class Device:
37+
device_type: str
38+
model_number: str
39+
product_manufacturer: str
40+
device_custom_info: dict[str, object] | None = None
41+
42+
43+
@dataclass(frozen=True)
44+
class DetectorControl:
45+
detection_type: str
46+
detection_duration_setting: float | None = None
47+
detector_relative_offset_setting: float | None = None
48+
detector_sampling_rate_setting: float | None = None
49+
m_z_maximum_setting: float | None = None
50+
m_z_minimum_setting: float | None = None
51+
polarity_setting: str | None = None
52+
53+
54+
@dataclass(frozen=True)
55+
class SampleIntroduction:
56+
sample_introduction_medium: str
57+
sample_introduction_mode_setting: str
58+
flow_rate_setting: float | None = None
59+
laser_firing_frequency_setting: float | None = None
60+
sample_introduction_description: str | None = None
61+
62+
63+
@dataclass(frozen=True)
64+
class Measurement:
65+
analyst: str
66+
submitter: str
67+
measurement_mode_setting: str
68+
69+
# Optional associations
70+
sample_identifier: str | None = None
71+
detector_control: DetectorControl | None = None
72+
sample_introduction: SampleIntroduction | None = None
73+
74+
# Optional aggregates
75+
processed_data_types: list[str] | None = None
76+
data_processing_types: list[str] | None = None
77+
78+
79+
@dataclass(frozen=True)
80+
class Peak:
81+
identifier: str | None = None
82+
m_z: float | None = None
83+
mass: float | None = None
84+
peak_area_value: float | None = None
85+
peak_area_unit: str | None = None
86+
peak_height_value: float | None = None
87+
peak_height_unit: str | None = None
88+
peak_width_value: float | None = None
89+
peak_width_unit: str | None = None
90+
relative_peak_area: float | None = None
91+
relative_peak_height: float | None = None
92+
written_name: str | None = None
93+
94+
95+
@dataclass(frozen=True)
96+
class Metadata:
97+
asset_management_identifier: str
98+
data_processing_description: str | None = None
99+
devices: list[Device] | None = None
100+
device_system_custom_info: dict[str, object] | None = None
101+
102+
103+
@dataclass(frozen=True)
104+
class Data:
105+
metadata: Metadata
106+
measurements: list[Measurement]
107+
peaks: list[Peak] | None = None
108+
109+
110+
class Mapper(SchemaMapper[Data, Model]):
111+
MANIFEST = "http://purl.allotrope.org/manifests/mass-spectrometry/REC/2025/06/mass-spectrometry.manifest"
112+
113+
def map_model(self, data: Data) -> Model:
114+
return Model(
115+
manifest=self.MANIFEST,
116+
data_processing_description=data.metadata.data_processing_description,
117+
mass_spectrometry_aggregate_document=MassSpectrometryAggregateDocument(
118+
device_system_document=self._get_device_system_document(data.metadata),
119+
mass_spectrometry_document=[
120+
self._get_mass_spectrometry_document_item(m)
121+
for m in data.measurements
122+
],
123+
),
124+
peak_list=self._get_peak_list(data.peaks),
125+
)
126+
127+
def _get_device_system_document(self, metadata: Metadata) -> DeviceSystemDocument:
128+
return add_custom_information_document(
129+
DeviceSystemDocument(
130+
asset_management_identifier=metadata.asset_management_identifier,
131+
device_document=[
132+
DeviceDocumentItem(
133+
device_type=device.device_type,
134+
model_number=device.model_number,
135+
product_manufacturer=device.product_manufacturer,
136+
)
137+
for device in (metadata.devices or [])
138+
]
139+
or None,
140+
),
141+
metadata.device_system_custom_info,
142+
)
143+
144+
def _get_mass_spectrometry_document_item(
145+
self, measurement: Measurement
146+
) -> MassSpectrometryDocumentItem:
147+
return MassSpectrometryDocumentItem(
148+
analyst=measurement.analyst,
149+
submitter=measurement.submitter,
150+
sample_document=(
151+
SampleDocument(sample_identifier=measurement.sample_identifier)
152+
if measurement.sample_identifier
153+
else None
154+
),
155+
sample_introduction_document=self._get_sample_introduction_document(
156+
measurement.sample_introduction
157+
),
158+
measurement_document=self._get_measurement_document(measurement),
159+
processed_data_aggregate_document=self._get_processed_data_aggregate_document(
160+
measurement.processed_data_types
161+
),
162+
data_processing_aggregate_document=self._get_data_processing_aggregate_document(
163+
measurement.data_processing_types
164+
),
165+
)
166+
167+
def _get_sample_introduction_document(
168+
self, intro: SampleIntroduction | None
169+
) -> SampleIntroductionDocument | None:
170+
if not intro:
171+
return None
172+
173+
return SampleIntroductionDocument(
174+
sample_introduction_medium=intro.sample_introduction_medium,
175+
sample_introduction_mode_setting=intro.sample_introduction_mode_setting,
176+
flow_rate_setting=quantity_or_none(
177+
TQuantityValueMilliliterPerMinute, intro.flow_rate_setting
178+
),
179+
laser_firing_frequency_setting=quantity_or_none(
180+
TQuantityValueHertz, intro.laser_firing_frequency_setting
181+
),
182+
sample_introduction_description=intro.sample_introduction_description,
183+
)
184+
185+
def _get_measurement_document(
186+
self, measurement: Measurement
187+
) -> MeasurementDocument:
188+
return MeasurementDocument(
189+
measurement_mode_setting=measurement.measurement_mode_setting,
190+
detector_control_aggregate_document=self._get_detector_control_aggregate_document(
191+
measurement.detector_control
192+
),
193+
)
194+
195+
def _get_detector_control_aggregate_document(
196+
self, control: DetectorControl | None
197+
) -> DetectorControlAggregateDocument | None:
198+
if not control:
199+
return None
200+
return DetectorControlAggregateDocument(
201+
detector_control_document=[
202+
DetectorControlDocumentItem(
203+
detection_type=control.detection_type,
204+
detection_duration_setting=quantity_or_none(
205+
TQuantityValueSecondTime, control.detection_duration_setting
206+
),
207+
detector_relative_offset_setting=quantity_or_none(
208+
TQuantityValueSecondTime,
209+
control.detector_relative_offset_setting,
210+
),
211+
detector_sampling_rate_setting=quantity_or_none(
212+
TQuantityValueHertz, control.detector_sampling_rate_setting
213+
),
214+
m_z_maximum_setting=quantity_or_none(
215+
TQuantityValueMassPerCharge, control.m_z_maximum_setting
216+
),
217+
m_z_minimum_setting=quantity_or_none(
218+
TQuantityValueMassPerCharge, control.m_z_minimum_setting
219+
),
220+
polarity_setting=control.polarity_setting,
221+
)
222+
]
223+
)
224+
225+
def _get_processed_data_aggregate_document(
226+
self, types: list[str] | None
227+
) -> ProcessedDataAggregateDocument | None:
228+
if not types:
229+
return None
230+
return ProcessedDataAggregateDocument(
231+
processed_data_document=[
232+
ProcessedDataDocumentItem(data_format_specification_type=t)
233+
for t in types
234+
]
235+
)
236+
237+
def _get_data_processing_aggregate_document(
238+
self, types: list[str] | None
239+
) -> DataProcessingAggregateDocument | None:
240+
if not types:
241+
return None
242+
return DataProcessingAggregateDocument(
243+
data_processing_document=[
244+
DataProcessingDocumentItem(data_processing_type=t) for t in types
245+
]
246+
)
247+
248+
def _get_peak_list(self, peaks: list[Peak] | None) -> PeakList | None:
249+
if not peaks:
250+
return None
251+
return PeakList(
252+
peak=[self._get_peak_item(p) for p in peaks] or None,
253+
)
254+
255+
def _get_peak_item(self, peak: Peak) -> PeakItem:
256+
return PeakItem(
257+
identifier=peak.identifier,
258+
m_z=quantity_or_none(TQuantityValueMassPerCharge, peak.m_z),
259+
mass=quantity_or_none(TQuantityValueDalton, peak.mass),
260+
peak_area=(
261+
TQuantityValue(value=peak.peak_area_value, unit=peak.peak_area_unit) # type: ignore[arg-type]
262+
if peak.peak_area_value is not None and peak.peak_area_unit is not None
263+
else None
264+
),
265+
peak_height=(
266+
TQuantityValue(value=peak.peak_height_value, unit=peak.peak_height_unit) # type: ignore[arg-type]
267+
if peak.peak_height_value is not None
268+
and peak.peak_height_unit is not None
269+
else None
270+
),
271+
peak_width=(
272+
TQuantityValue(value=peak.peak_width_value, unit=peak.peak_width_unit) # type: ignore[arg-type]
273+
if peak.peak_width_value is not None
274+
and peak.peak_width_unit is not None
275+
else None
276+
),
277+
relative_peak_area=quantity_or_none(
278+
TQuantityValuePercent, peak.relative_peak_area
279+
),
280+
relative_peak_height=quantity_or_none(
281+
TQuantityValuePercent, peak.relative_peak_height
282+
),
283+
written_name=peak.written_name,
284+
)

0 commit comments

Comments
 (0)