From 611aa645ecd9a613c5964e93cd45c6e02be3f8b0 Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Thu, 13 Nov 2025 12:15:30 -0500 Subject: [PATCH 1/6] Added mapper and test for it --- .../rec/_2025/_06/mass_spectrometry.py | 284 ++++++++++++++++++ ...ass_spectrometry_mapper_dummy_data_test.py | 145 +++++++++ 2 files changed, 429 insertions(+) create mode 100644 src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py create mode 100644 tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py diff --git a/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py b/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py new file mode 100644 index 000000000..f106bbd30 --- /dev/null +++ b/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py @@ -0,0 +1,284 @@ +from dataclasses import dataclass + +from allotropy.allotrope.converter import add_custom_information_document +from allotropy.allotrope.models.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import ( + DataProcessingAggregateDocument, + DataProcessingDocumentItem, + DetectorControlAggregateDocument, + DetectorControlDocumentItem, + DeviceDocumentItem, + DeviceSystemDocument, + MassSpectrometryAggregateDocument, + MassSpectrometryDocumentItem, + MeasurementDocument, + Model, + PeakItem, + PeakList, + ProcessedDataAggregateDocument, + ProcessedDataDocumentItem, + SampleDocument, + SampleIntroductionDocument, +) +from allotropy.allotrope.models.shared.definitions.custom import ( + TQuantityValueDalton, + TQuantityValueHertz, + TQuantityValueMassPerCharge, + TQuantityValueMilliliterPerMinute, + TQuantityValuePercent, + TQuantityValueSecondTime, +) +from allotropy.allotrope.models.shared.definitions.definitions import TQuantityValue +from allotropy.allotrope.schema_mappers.schema_mapper import SchemaMapper +from allotropy.parsers.utils.values import quantity_or_none + + +@dataclass(frozen=True) +class Device: + device_type: str + model_number: str + product_manufacturer: str + device_custom_info: dict[str, object] | None = None + + +@dataclass(frozen=True) +class DetectorControl: + detection_type: str + detection_duration_setting: float | None = None + detector_relative_offset_setting: float | None = None + detector_sampling_rate_setting: float | None = None + m_z_maximum_setting: float | None = None + m_z_minimum_setting: float | None = None + polarity_setting: str | None = None + + +@dataclass(frozen=True) +class SampleIntroduction: + sample_introduction_medium: str + sample_introduction_mode_setting: str + flow_rate_setting: float | None = None + laser_firing_frequency_setting: float | None = None + sample_introduction_description: str | None = None + + +@dataclass(frozen=True) +class Measurement: + analyst: str + submitter: str + measurement_mode_setting: str + + # Optional associations + sample_identifier: str | None = None + detector_control: DetectorControl | None = None + sample_introduction: SampleIntroduction | None = None + + # Optional aggregates + processed_data_types: list[str] | None = None + data_processing_types: list[str] | None = None + + +@dataclass(frozen=True) +class Peak: + identifier: str | None = None + m_z: float | None = None + mass: float | None = None + peak_area_value: float | None = None + peak_area_unit: str | None = None + peak_height_value: float | None = None + peak_height_unit: str | None = None + peak_width_value: float | None = None + peak_width_unit: str | None = None + relative_peak_area: float | None = None + relative_peak_height: float | None = None + written_name: str | None = None + + +@dataclass(frozen=True) +class Metadata: + asset_management_identifier: str + data_processing_description: str | None = None + devices: list[Device] | None = None + device_system_custom_info: dict[str, object] | None = None + + +@dataclass(frozen=True) +class Data: + metadata: Metadata + measurements: list[Measurement] + peaks: list[Peak] | None = None + + +class Mapper(SchemaMapper[Data, Model]): + MANIFEST = "http://purl.allotrope.org/manifests/mass-spectrometry/REC/2025/06/mass-spectrometry.manifest" + + def map_model(self, data: Data) -> Model: + return Model( + manifest=self.MANIFEST, + data_processing_description=data.metadata.data_processing_description, + mass_spectrometry_aggregate_document=MassSpectrometryAggregateDocument( + device_system_document=self._get_device_system_document(data.metadata), + mass_spectrometry_document=[ + self._get_mass_spectrometry_document_item(m) + for m in data.measurements + ], + ), + peak_list=self._get_peak_list(data.peaks), + ) + + def _get_device_system_document(self, metadata: Metadata) -> DeviceSystemDocument: + return add_custom_information_document( + DeviceSystemDocument( + asset_management_identifier=metadata.asset_management_identifier, + device_document=[ + DeviceDocumentItem( + device_type=device.device_type, + model_number=device.model_number, + product_manufacturer=device.product_manufacturer, + ) + for device in (metadata.devices or []) + ] + or None, + ), + metadata.device_system_custom_info, + ) + + def _get_mass_spectrometry_document_item( + self, measurement: Measurement + ) -> MassSpectrometryDocumentItem: + return MassSpectrometryDocumentItem( + analyst=measurement.analyst, + submitter=measurement.submitter, + sample_document=( + SampleDocument(sample_identifier=measurement.sample_identifier) + if measurement.sample_identifier + else None + ), + sample_introduction_document=self._get_sample_introduction_document( + measurement.sample_introduction + ), + measurement_document=self._get_measurement_document(measurement), + processed_data_aggregate_document=self._get_processed_data_aggregate_document( + measurement.processed_data_types + ), + data_processing_aggregate_document=self._get_data_processing_aggregate_document( + measurement.data_processing_types + ), + ) + + def _get_sample_introduction_document( + self, intro: SampleIntroduction | None + ) -> SampleIntroductionDocument | None: + if not intro: + return None + + return SampleIntroductionDocument( + sample_introduction_medium=intro.sample_introduction_medium, + sample_introduction_mode_setting=intro.sample_introduction_mode_setting, + flow_rate_setting=quantity_or_none( + TQuantityValueMilliliterPerMinute, intro.flow_rate_setting + ), + laser_firing_frequency_setting=quantity_or_none( + TQuantityValueHertz, intro.laser_firing_frequency_setting + ), + sample_introduction_description=intro.sample_introduction_description, + ) + + def _get_measurement_document( + self, measurement: Measurement + ) -> MeasurementDocument: + return MeasurementDocument( + measurement_mode_setting=measurement.measurement_mode_setting, + detector_control_aggregate_document=self._get_detector_control_aggregate_document( + measurement.detector_control + ), + ) + + def _get_detector_control_aggregate_document( + self, control: DetectorControl | None + ) -> DetectorControlAggregateDocument | None: + if not control: + return None + return DetectorControlAggregateDocument( + detector_control_document=[ + DetectorControlDocumentItem( + detection_type=control.detection_type, + detection_duration_setting=quantity_or_none( + TQuantityValueSecondTime, control.detection_duration_setting + ), + detector_relative_offset_setting=quantity_or_none( + TQuantityValueSecondTime, + control.detector_relative_offset_setting, + ), + detector_sampling_rate_setting=quantity_or_none( + TQuantityValueHertz, control.detector_sampling_rate_setting + ), + m_z_maximum_setting=quantity_or_none( + TQuantityValueMassPerCharge, control.m_z_maximum_setting + ), + m_z_minimum_setting=quantity_or_none( + TQuantityValueMassPerCharge, control.m_z_minimum_setting + ), + polarity_setting=control.polarity_setting, + ) + ] + ) + + def _get_processed_data_aggregate_document( + self, types: list[str] | None + ) -> ProcessedDataAggregateDocument | None: + if not types: + return None + return ProcessedDataAggregateDocument( + processed_data_document=[ + ProcessedDataDocumentItem(data_format_specification_type=t) + for t in types + ] + ) + + def _get_data_processing_aggregate_document( + self, types: list[str] | None + ) -> DataProcessingAggregateDocument | None: + if not types: + return None + return DataProcessingAggregateDocument( + data_processing_document=[ + DataProcessingDocumentItem(data_processing_type=t) for t in types + ] + ) + + def _get_peak_list(self, peaks: list[Peak] | None) -> PeakList | None: + if not peaks: + return None + return PeakList( + peak=[self._get_peak_item(p) for p in peaks] or None, + ) + + def _get_peak_item(self, peak: Peak) -> PeakItem: + return PeakItem( + identifier=peak.identifier, + m_z=quantity_or_none(TQuantityValueMassPerCharge, peak.m_z), + mass=quantity_or_none(TQuantityValueDalton, peak.mass), + peak_area=( + TQuantityValue(value=peak.peak_area_value, unit=peak.peak_area_unit) # type: ignore[arg-type] + if peak.peak_area_value is not None and peak.peak_area_unit is not None + else None + ), + peak_height=( + TQuantityValue(value=peak.peak_height_value, unit=peak.peak_height_unit) # type: ignore[arg-type] + if peak.peak_height_value is not None + and peak.peak_height_unit is not None + else None + ), + peak_width=( + TQuantityValue(value=peak.peak_width_value, unit=peak.peak_width_unit) # type: ignore[arg-type] + if peak.peak_width_value is not None + and peak.peak_width_unit is not None + else None + ), + relative_peak_area=quantity_or_none( + TQuantityValuePercent, peak.relative_peak_area + ), + relative_peak_height=quantity_or_none( + TQuantityValuePercent, peak.relative_peak_height + ), + written_name=peak.written_name, + ) diff --git a/tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py b/tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py new file mode 100644 index 000000000..efd92347a --- /dev/null +++ b/tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +from allotropy.allotrope.allotrope import serialize_and_validate_allotrope +from allotropy.allotrope.schema_mappers.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import ( + Data, + DetectorControl, + Device, + Mapper, + Measurement, + Metadata, + Peak, + SampleIntroduction, +) +from allotropy.parsers.utils.timestamp_parser import TimestampParser + + +def build_dummy_data() -> Data: + return Data( + metadata=Metadata( + asset_management_identifier="asset-management-identifier-001", + data_processing_description="example data processing description", + devices=[ + Device( + device_type="spectrometer", + model_number="example model number", + product_manufacturer="example product manufacturer", + device_custom_info={ + "example device custom info key": "example device custom info value" + }, + ), + Device( + device_type="spectrometer", + model_number="example model number", + product_manufacturer="example product manufacturer", + device_custom_info={ + "example device custom info key": "example device custom info value" + }, + ), + ], + device_system_custom_info={ + "example device system custom info key": "example device system custom info value" + }, + ), + measurements=[ + Measurement( + analyst="example analyst", + submitter="example submitter", + measurement_mode_setting="example measurement mode setting", + sample_identifier="sample-identifier-001", + detector_control=DetectorControl( + detection_type="example detection type", + detection_duration_setting=1.0, + detector_relative_offset_setting=1.0, + detector_sampling_rate_setting=1.0, + m_z_maximum_setting=1.0, + m_z_minimum_setting=1.0, + polarity_setting="example polarity setting", + ), + sample_introduction=SampleIntroduction( + sample_introduction_medium="example sample introduction medium", + sample_introduction_mode_setting="example sample introduction mode setting", + flow_rate_setting=1.0, + laser_firing_frequency_setting=1.0, + sample_introduction_description="example sample introduction description", + ), + processed_data_types=[ + "example processed data types", + "example processed data types", + ], + data_processing_types=[ + "example data processing types", + "example data processing types", + ], + ), + Measurement( + analyst="example analyst", + submitter="example submitter", + measurement_mode_setting="example measurement mode setting", + sample_identifier="sample-identifier-002", + detector_control=DetectorControl( + detection_type="example detection type", + detection_duration_setting=1.0, + detector_relative_offset_setting=1.0, + detector_sampling_rate_setting=1.0, + m_z_maximum_setting=1.0, + m_z_minimum_setting=1.0, + polarity_setting="example polarity setting", + ), + sample_introduction=SampleIntroduction( + sample_introduction_medium="example sample introduction medium", + sample_introduction_mode_setting="example sample introduction mode setting", + flow_rate_setting=1.0, + laser_firing_frequency_setting=1.0, + sample_introduction_description="example sample introduction description", + ), + processed_data_types=[ + "example processed data types", + "example processed data types", + ], + data_processing_types=[ + "example data processing types", + "example data processing types", + ], + ), + ], + peaks=[ + Peak( + identifier="identifier-001", + m_z=1.0, + mass=1.0, + peak_area_value=1.0, + peak_area_unit="example peak area unit", + peak_height_value=1.0, + peak_height_unit="example peak height unit", + peak_width_value=1.0, + peak_width_unit="example peak width unit", + relative_peak_area=1.0, + relative_peak_height=1.0, + written_name="example written name", + ), + Peak( + identifier="identifier-002", + m_z=1.0, + mass=1.0, + peak_area_value=1.0, + peak_area_unit="example peak area unit", + peak_height_value=1.0, + peak_height_unit="example peak height unit", + peak_width_value=1.0, + peak_width_unit="example peak width unit", + relative_peak_area=1.0, + relative_peak_height=1.0, + written_name="example written name", + ), + ], + ) + + +def test_mass_spectrometry_mapper_dummy_data() -> None: + # It just validates that the mapper and serializer can handle the dummy data without raising an exception + data = build_dummy_data() + mapper_output = Mapper( + "asm_converter_name", lambda time: TimestampParser().parse(time) + ).map_model(data) + serialize_and_validate_allotrope(mapper_output) From ceadf41ca7a22551f314cc3be5b32b45d9ad5b5e Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Thu, 13 Nov 2025 12:28:59 -0500 Subject: [PATCH 2/6] helper scripts --- scripts/generate_dummy_data_from_mapper.py | 258 ++++++++++++++++++ .../generate_fake_json_from_dataclasses.py | 248 +++++++++++++++++ .../rec/_2025/_06/mass_spectrometry.py | 6 +- 3 files changed, 509 insertions(+), 3 deletions(-) create mode 100644 scripts/generate_dummy_data_from_mapper.py create mode 100644 scripts/generate_fake_json_from_dataclasses.py diff --git a/scripts/generate_dummy_data_from_mapper.py b/scripts/generate_dummy_data_from_mapper.py new file mode 100644 index 000000000..7f0db2aed --- /dev/null +++ b/scripts/generate_dummy_data_from_mapper.py @@ -0,0 +1,258 @@ +# ruff: noqa: S603 +from __future__ import annotations + +import argparse +import dataclasses +import importlib.util +import inspect +from pathlib import Path +import subprocess +import sys +import types +import typing +from typing import Any, get_args, get_origin + + +def _module_from_path(module_path: Path) -> types.ModuleType: + spec = importlib.util.spec_from_file_location(module_path.stem, str(module_path)) + if spec is None or spec.loader is None: + msg = f"Unable to load module from {module_path}" + raise RuntimeError(msg) + module = importlib.util.module_from_spec(spec) + sys.modules[module_path.stem] = module + spec.loader.exec_module(module) + return module + + +def _compute_dotted_import(module_path: Path) -> str: + # Try to compute dotted path relative to repo's src directory + repo_root = Path(__file__).resolve().parents[1] + src_dir = repo_root / "src" + try: + rel = module_path.resolve().relative_to(src_dir.resolve()) + dotted = ".".join(rel.with_suffix("").parts) + return dotted + except Exception: + # Fallback: use module name only (may fail if not on sys.path) + return module_path.stem + + +def _is_dataclass_type(tp: Any) -> bool: + try: + return inspect.isclass(tp) and dataclasses.is_dataclass(tp) + except Exception: + return False + + +def _unwrap_optional(tp: Any) -> Any: + origin = get_origin(tp) + if origin is None: + return tp + if origin in (list, dict, tuple): + return tp + if ( + origin is typing.Union + or str(origin) == "typing.Union" + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) + ): + args = [a for a in get_args(tp) if a is not type(None)] + return args[0] if args else tp + return tp + + +def _choose_union_member(tp: Any) -> Any: + origin = get_origin(tp) + if origin is None: + return tp + if ( + origin is typing.Union + or str(origin) == "typing.Union" + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) + ): + args = list(get_args(tp)) + for preferred in (str, int, float, bool): + if preferred in args: + return preferred + for arg in args: + if arg is not type(None): + return arg + return tp + + +def _collect_dataclass_types( + cls: type[Any], seen: set[type[Any]] | None = None +) -> set[type[Any]]: + if seen is None: + seen = set() + if cls in seen: + return seen + seen.add(cls) + try: + type_hints = typing.get_type_hints(cls, include_extras=True) + except Exception: + type_hints = {} + for f in dataclasses.fields(cls): + tp = type_hints.get(f.name, f.type) + tp = _choose_union_member(tp) + tp = _unwrap_optional(tp) + origin = get_origin(tp) + args = get_args(tp) + if origin is list and args: + inner = _unwrap_optional(_choose_union_member(args[0])) + if _is_dataclass_type(inner): + _collect_dataclass_types(inner, seen) + elif _is_dataclass_type(tp): + _collect_dataclass_types(tp, seen) + return seen + + +def _py_literal_for_scalar(tp: Any, field_name: str, list_index: int | None) -> str: + lname = field_name.lower() + if "identifier" in lname: + suffix = (list_index + 1) if isinstance(list_index, int) else 1 + base = field_name.replace("_", "-").lower() + return f'"{base}-{suffix:03d}"' + if tp is str: + return f'"example {field_name.replace("_", " ")}"' + if tp is int: + return "1" + if tp is float: + return "1.0" + if tp is bool: + return "True" + return f'"example {field_name.replace("_", " ")}"' + + +def _render_value_expr( + tp: Any, field_name: str, indent: int, list_index: int | None +) -> str: + tp = _choose_union_member(tp) + tp = _unwrap_optional(tp) + origin = get_origin(tp) + args = get_args(tp) + sp = " " * indent + # Dicts: generate one representative key/value pair + if origin is dict and args: + _, val_tp = args[0], args[1] + key_expr = _py_literal_for_scalar(str, f"{field_name}_key", list_index) + val_expr = _render_value_expr( + val_tp, f"{field_name}_value", indent + 4, list_index + ) + return "{\n" + f"{sp} {key_expr}: {val_expr}\n" + f"{sp}" + "}" + if origin is list and args: + inner = args[0] + items = [] + for i in range(2): + items.append(_render_value_expr(inner, field_name, indent + 4, i)) + # Determine if items are multiline (start with a class constructor or open bracket) + multiline = any( + x.strip().startswith(tuple("ABCDEFGHIJKLMNOPQRSTUVWXYZ")) + or x.strip().startswith("[") + or "\n" in x + for x in items + ) + if multiline: + inner_joined = ",\n".join(items) + return f"[\n{inner_joined}\n{sp}]" + return f"[{', '.join(items)}]" + if _is_dataclass_type(tp): + return _render_dataclass_expr(tp, indent, list_index) + # scalar + chosen = _choose_union_member(tp) + chosen = _unwrap_optional(chosen) + return _py_literal_for_scalar(chosen, field_name, list_index) + + +def _render_dataclass_expr(cls: type[Any], indent: int, list_index: int | None) -> str: + sp = " " * indent + lines = [f"{sp}{cls.__name__}("] + try: + module_globals = sys.modules.get(cls.__module__).__dict__ + except Exception: + module_globals = None + try: + type_hints = typing.get_type_hints( + cls, globalns=module_globals, localns=module_globals, include_extras=True + ) + except Exception: + type_hints = {} + for f in dataclasses.fields(cls): + key = f.name + tp = type_hints.get(f.name, f.type) + val_expr = _render_value_expr(tp, key, indent + 4, list_index) + lines.append(f"{sp} {key}={val_expr},") + lines.append(f"{sp})") + return "\n".join(lines) + + +def generate_dummy_data_py( + mapper_file: Path, top_class_name: str, output: Path +) -> None: + # Ensure repo 'src' is on sys.path for module imports + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + module = _module_from_path(mapper_file) + + top_cls = getattr(module, top_class_name, None) + if top_cls is None or not _is_dataclass_type(top_cls): + msg = f"Dataclass '{top_class_name}' not found in {mapper_file}" + raise RuntimeError(msg) + + # Collect all dataclasses used so we can import them in the generated file + dataclass_types = _collect_dataclass_types(top_cls) + # Ensure top class first in import list, followed by others sorted + class_names = [ + top_cls.__name__, + *sorted([c.__name__ for c in dataclass_types if c is not top_cls]), + ] + + dotted_import = _compute_dotted_import(mapper_file) + header = f"""from __future__ import annotations + +from dataclasses import asdict +import json + +from {dotted_import} import ( + {", ".join(class_names)} +) + + +def build_dummy_data() -> {top_cls.__name__}: +""" + body_expr = _render_dataclass_expr(top_cls, indent=4, list_index=None) + + content = header + " return " + body_expr + + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(content, encoding="utf-8") + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Generate a Python file that builds dummy Data-like object for any mapper." + ) + parser.add_argument( + "--module-file", + required=True, + type=Path, + help="Path to the mapper module file (e.g., schema_mappers/.../mass_spectrometry.py).", + ) + parser.add_argument( + "--output", + required=True, + type=Path, + help="Path to write the generated Python file.", + ) + parser.add_argument( + "--top-class", + default="Data", + help="Top-level dataclass name to start from. Defaults to 'Data'.", + ) + args = parser.parse_args() + generate_dummy_data_py(args.module_file, args.top_class, args.output) + print(f"Wrote dummy data generator to: {args.output}") + output_path = str(args.output) + subprocess.run(["ruff", "check", "--fix", output_path], check=True) # noqa: S607 + + +if __name__ == "__main__": + main() diff --git a/scripts/generate_fake_json_from_dataclasses.py b/scripts/generate_fake_json_from_dataclasses.py new file mode 100644 index 000000000..f659b3391 --- /dev/null +++ b/scripts/generate_fake_json_from_dataclasses.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import argparse +import dataclasses +import importlib.util +import inspect +import json +import logging +from pathlib import Path +import sys +import types +from types import ModuleType +import typing +from typing import Any, get_args, get_origin + + +def _to_space_separated(name: str) -> str: + return name.replace("_", " ").strip().lower() + + +def _load_module_from_path(module_path: Path) -> ModuleType: + spec = importlib.util.spec_from_file_location(module_path.stem, str(module_path)) + if spec is None or spec.loader is None: + msg = f"Unable to load module from {module_path}" + raise RuntimeError(msg) + module = importlib.util.module_from_spec(spec) + sys.modules[module_path.stem] = module + spec.loader.exec_module(module) + return module + + +def _is_dataclass_type(tp: Any) -> bool: + try: + return inspect.isclass(tp) and dataclasses.is_dataclass(tp) + except Exception: + return False + + +def _unwrap_optional(tp: Any) -> Any: + origin = get_origin(tp) + if origin is None: + return tp + if origin is list or origin is dict or origin is tuple: + return tp + if ( + origin is typing.Union + or str(origin) == "typing.Union" + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) + ): + args = [a for a in get_args(tp) if a is not type(None)] + return args[0] if args else tp + return tp + + +def _choose_union_member(tp: Any) -> Any: + origin = get_origin(tp) + if origin is None: + return tp + if ( + origin is typing.Union + or str(origin) == "typing.Union" + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) + ): + args = list(get_args(tp)) + # Prefer primitives over dataclasses for friendly fake data + for preferred in (str, int, float, bool): + if preferred in args: + return preferred + for arg in args: + if arg is not type(None): + return arg + return tp + + +def _unit_string_from_quantity_type(tp: Any) -> str | None: + # Quantity types in this codebase subclass a Units class from + # allotropy.allotrope.models.shared.definitions.units which has a default 'unit' value. + try: + for base in getattr(tp, "__mro__", []): + # Skip the quantity base itself; we need the units mixin + if base is tp: + continue + # Heuristic: a dataclass with a 'unit' field default on the class + if dataclasses.is_dataclass(base) and any( + f.name == "unit" for f in dataclasses.fields(base) + ): + # Try to read class attribute default, else instantiate + val = getattr(base, "unit", None) + if isinstance(val, str) and val: + return val + try: + instance = base() + val2 = getattr(instance, "unit", None) + if isinstance(val2, str) and val2: + return val2 + except Exception as exc: + logging.debug("Ignoring unit mixin instantiation error: %s", exc) + continue + return None + except Exception: + return None + + +def _fake_scalar_for_type( + tp: Any, field_name: str, ctx: dict[str, Any] | None = None +) -> Any: + lname = field_name.lower() + if "identifier" in lname: + base = field_name.replace("_", "-").lower() + idx = None if ctx is None else ctx.get("list_index") + if isinstance(idx, int): + return f"{base}-{idx + 1:03d}" + return f"{base}-001" + # TStringValue, TClass, TNamed → str (these are unions including str) + if tp in (str,): + # Make the fake a bit readable from the field context + return f"example {field_name.replace('_', ' ')}" + if tp in (int,): + return 1 + if tp in (float,): + return 1.0 + if tp in (bool,): + return True + return f"example {field_name.replace('_', ' ')}" + + +def _fake_for_type(tp: Any, field_name: str, ctx: dict[str, Any] | None = None) -> Any: + # Resolve Optional/Union + tp = _choose_union_member(tp) + tp = _unwrap_optional(tp) + + origin = get_origin(tp) + args = get_args(tp) + + # Lists + if origin is list and args: + inner = args[0] + items = [] + for i in range(2): + child_ctx = {} if ctx is None else dict(ctx) + child_ctx["list_index"] = i + # If the inner is a dataclass, delegate to dataclass builder to ensure nested structure + if _is_dataclass_type(_unwrap_optional(_choose_union_member(inner))): + items.append( + _build_object_for_dataclass( + _unwrap_optional(_choose_union_member(inner)), child_ctx + ) + ) + else: + items.append(_fake_for_type(inner, field_name, child_ctx)) + return items + + # Dataclasses (including nested models and quantity values) + if _is_dataclass_type(tp): + # Quantity-like dataclasses: look for 'value' and 'unit' fields (from TQuantityValue) + try: + field_names = [f.name for f in dataclasses.fields(tp)] + if "value" in field_names and "unit" in field_names: + unit = _unit_string_from_quantity_type(tp) or "unit" + return {"value": 1.0, "unit": unit} + except Exception as exc: + logging.debug("Dataclass field inspection failed: %s", exc) + # Generic nested dataclass: return its object (without wrapping in the class name) + return _build_object_for_dataclass(tp, ctx) + + # Parameterized types we don't explicitly handle: fall back + if origin is not None: + return f"example {field_name.replace('_', ' ')}" + + # Primitive or aliases that resolve to primitives + return _fake_scalar_for_type(tp, field_name, ctx) + + +def _build_object_for_dataclass( + cls: type[Any], ctx: dict[str, Any] | None = None +) -> dict[str, Any]: + result: dict[str, Any] = {} + # Resolve postponed annotations so we get real types instead of strings + try: + module_globals = sys.modules.get(cls.__module__).__dict__ + except Exception: + module_globals = None + try: + # include_extras preserves typing like list[Foo] etc. + type_hints = typing.get_type_hints( + cls, globalns=module_globals, localns=module_globals, include_extras=True + ) + except Exception: + type_hints = {} + for f in dataclasses.fields(cls): + key = _to_space_separated(f.name) + # If there is a non-None default (e.g., manifest), keep it + if f.default is not dataclasses.MISSING and f.default is not None: + result[key] = f.default + continue + annotated_type = type_hints.get(f.name, f.type) + result[key] = _fake_for_type(annotated_type, f.name, ctx) + return result + + +def generate_fake_json( + module_file: Path, top_class_name: str = "Model" +) -> dict[str, Any]: + # Ensure repo 'src' is on sys.path so imports inside the model module work + src_dir = Path(__file__).resolve().parents[1] / "src" + sys.path.insert(0, str(src_dir)) + + module = _load_module_from_path(module_file) + model_cls = getattr(module, top_class_name, None) + if model_cls is None or not _is_dataclass_type(model_cls): + msg = f"Dataclass '{top_class_name}' not found in {module_file}" + raise RuntimeError(msg) + payload = _build_object_for_dataclass(model_cls) + # Return the model's fields at the top level (no top-level 'model' key) + return payload + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Generate a fake JSON document from a dataclasses module (top class 'Model')." + ) + parser.add_argument( + "--module-file", + required=True, + type=Path, + help="Path to the Python module containing the generated dataclasses (e.g., mass_spectrometry.py)", + ) + parser.add_argument( + "--output", + required=True, + type=Path, + help="Path to write the resulting JSON file.", + ) + parser.add_argument( + "--top-class", + default="Model", + help="Top-level dataclass name to start from. Defaults to 'Model'.", + ) + args = parser.parse_args() + + data = generate_fake_json(args.module_file, args.top_class) + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", encoding="utf-8") as f: + json.dump(data, f, indent=4, ensure_ascii=False) + + +if __name__ == "__main__": + main() diff --git a/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py b/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py index f106bbd30..5f5fca6fc 100644 --- a/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py +++ b/src/allotropy/allotrope/schema_mappers/adm/mass_spectrometry/rec/_2025/_06/mass_spectrometry.py @@ -258,18 +258,18 @@ def _get_peak_item(self, peak: Peak) -> PeakItem: m_z=quantity_or_none(TQuantityValueMassPerCharge, peak.m_z), mass=quantity_or_none(TQuantityValueDalton, peak.mass), peak_area=( - TQuantityValue(value=peak.peak_area_value, unit=peak.peak_area_unit) # type: ignore[arg-type] + TQuantityValue(value=peak.peak_area_value, unit=peak.peak_area_unit) if peak.peak_area_value is not None and peak.peak_area_unit is not None else None ), peak_height=( - TQuantityValue(value=peak.peak_height_value, unit=peak.peak_height_unit) # type: ignore[arg-type] + TQuantityValue(value=peak.peak_height_value, unit=peak.peak_height_unit) if peak.peak_height_value is not None and peak.peak_height_unit is not None else None ), peak_width=( - TQuantityValue(value=peak.peak_width_value, unit=peak.peak_width_unit) # type: ignore[arg-type] + TQuantityValue(value=peak.peak_width_value, unit=peak.peak_width_unit) if peak.peak_width_value is not None and peak.peak_width_unit is not None else None From e9d4b9ceb4773d95ce44175a220bf569e351bf22 Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Fri, 21 Nov 2025 12:02:52 -0500 Subject: [PATCH 3/6] Added changes requested by comments --- scripts/generate_dummy_data_from_mapper.py | 258 ------------------ .../generate_fake_json_from_dataclasses.py | 248 ----------------- ...ass_spectrometry_mapper_dummy_data_test.py | 2 - 3 files changed, 508 deletions(-) delete mode 100644 scripts/generate_dummy_data_from_mapper.py delete mode 100644 scripts/generate_fake_json_from_dataclasses.py rename tests/allotrope/schema_parser/{ => mass_spectrometry}/mass_spectrometry_mapper_dummy_data_test.py (99%) diff --git a/scripts/generate_dummy_data_from_mapper.py b/scripts/generate_dummy_data_from_mapper.py deleted file mode 100644 index 7f0db2aed..000000000 --- a/scripts/generate_dummy_data_from_mapper.py +++ /dev/null @@ -1,258 +0,0 @@ -# ruff: noqa: S603 -from __future__ import annotations - -import argparse -import dataclasses -import importlib.util -import inspect -from pathlib import Path -import subprocess -import sys -import types -import typing -from typing import Any, get_args, get_origin - - -def _module_from_path(module_path: Path) -> types.ModuleType: - spec = importlib.util.spec_from_file_location(module_path.stem, str(module_path)) - if spec is None or spec.loader is None: - msg = f"Unable to load module from {module_path}" - raise RuntimeError(msg) - module = importlib.util.module_from_spec(spec) - sys.modules[module_path.stem] = module - spec.loader.exec_module(module) - return module - - -def _compute_dotted_import(module_path: Path) -> str: - # Try to compute dotted path relative to repo's src directory - repo_root = Path(__file__).resolve().parents[1] - src_dir = repo_root / "src" - try: - rel = module_path.resolve().relative_to(src_dir.resolve()) - dotted = ".".join(rel.with_suffix("").parts) - return dotted - except Exception: - # Fallback: use module name only (may fail if not on sys.path) - return module_path.stem - - -def _is_dataclass_type(tp: Any) -> bool: - try: - return inspect.isclass(tp) and dataclasses.is_dataclass(tp) - except Exception: - return False - - -def _unwrap_optional(tp: Any) -> Any: - origin = get_origin(tp) - if origin is None: - return tp - if origin in (list, dict, tuple): - return tp - if ( - origin is typing.Union - or str(origin) == "typing.Union" - or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) - ): - args = [a for a in get_args(tp) if a is not type(None)] - return args[0] if args else tp - return tp - - -def _choose_union_member(tp: Any) -> Any: - origin = get_origin(tp) - if origin is None: - return tp - if ( - origin is typing.Union - or str(origin) == "typing.Union" - or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) - ): - args = list(get_args(tp)) - for preferred in (str, int, float, bool): - if preferred in args: - return preferred - for arg in args: - if arg is not type(None): - return arg - return tp - - -def _collect_dataclass_types( - cls: type[Any], seen: set[type[Any]] | None = None -) -> set[type[Any]]: - if seen is None: - seen = set() - if cls in seen: - return seen - seen.add(cls) - try: - type_hints = typing.get_type_hints(cls, include_extras=True) - except Exception: - type_hints = {} - for f in dataclasses.fields(cls): - tp = type_hints.get(f.name, f.type) - tp = _choose_union_member(tp) - tp = _unwrap_optional(tp) - origin = get_origin(tp) - args = get_args(tp) - if origin is list and args: - inner = _unwrap_optional(_choose_union_member(args[0])) - if _is_dataclass_type(inner): - _collect_dataclass_types(inner, seen) - elif _is_dataclass_type(tp): - _collect_dataclass_types(tp, seen) - return seen - - -def _py_literal_for_scalar(tp: Any, field_name: str, list_index: int | None) -> str: - lname = field_name.lower() - if "identifier" in lname: - suffix = (list_index + 1) if isinstance(list_index, int) else 1 - base = field_name.replace("_", "-").lower() - return f'"{base}-{suffix:03d}"' - if tp is str: - return f'"example {field_name.replace("_", " ")}"' - if tp is int: - return "1" - if tp is float: - return "1.0" - if tp is bool: - return "True" - return f'"example {field_name.replace("_", " ")}"' - - -def _render_value_expr( - tp: Any, field_name: str, indent: int, list_index: int | None -) -> str: - tp = _choose_union_member(tp) - tp = _unwrap_optional(tp) - origin = get_origin(tp) - args = get_args(tp) - sp = " " * indent - # Dicts: generate one representative key/value pair - if origin is dict and args: - _, val_tp = args[0], args[1] - key_expr = _py_literal_for_scalar(str, f"{field_name}_key", list_index) - val_expr = _render_value_expr( - val_tp, f"{field_name}_value", indent + 4, list_index - ) - return "{\n" + f"{sp} {key_expr}: {val_expr}\n" + f"{sp}" + "}" - if origin is list and args: - inner = args[0] - items = [] - for i in range(2): - items.append(_render_value_expr(inner, field_name, indent + 4, i)) - # Determine if items are multiline (start with a class constructor or open bracket) - multiline = any( - x.strip().startswith(tuple("ABCDEFGHIJKLMNOPQRSTUVWXYZ")) - or x.strip().startswith("[") - or "\n" in x - for x in items - ) - if multiline: - inner_joined = ",\n".join(items) - return f"[\n{inner_joined}\n{sp}]" - return f"[{', '.join(items)}]" - if _is_dataclass_type(tp): - return _render_dataclass_expr(tp, indent, list_index) - # scalar - chosen = _choose_union_member(tp) - chosen = _unwrap_optional(chosen) - return _py_literal_for_scalar(chosen, field_name, list_index) - - -def _render_dataclass_expr(cls: type[Any], indent: int, list_index: int | None) -> str: - sp = " " * indent - lines = [f"{sp}{cls.__name__}("] - try: - module_globals = sys.modules.get(cls.__module__).__dict__ - except Exception: - module_globals = None - try: - type_hints = typing.get_type_hints( - cls, globalns=module_globals, localns=module_globals, include_extras=True - ) - except Exception: - type_hints = {} - for f in dataclasses.fields(cls): - key = f.name - tp = type_hints.get(f.name, f.type) - val_expr = _render_value_expr(tp, key, indent + 4, list_index) - lines.append(f"{sp} {key}={val_expr},") - lines.append(f"{sp})") - return "\n".join(lines) - - -def generate_dummy_data_py( - mapper_file: Path, top_class_name: str, output: Path -) -> None: - # Ensure repo 'src' is on sys.path for module imports - sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) - module = _module_from_path(mapper_file) - - top_cls = getattr(module, top_class_name, None) - if top_cls is None or not _is_dataclass_type(top_cls): - msg = f"Dataclass '{top_class_name}' not found in {mapper_file}" - raise RuntimeError(msg) - - # Collect all dataclasses used so we can import them in the generated file - dataclass_types = _collect_dataclass_types(top_cls) - # Ensure top class first in import list, followed by others sorted - class_names = [ - top_cls.__name__, - *sorted([c.__name__ for c in dataclass_types if c is not top_cls]), - ] - - dotted_import = _compute_dotted_import(mapper_file) - header = f"""from __future__ import annotations - -from dataclasses import asdict -import json - -from {dotted_import} import ( - {", ".join(class_names)} -) - - -def build_dummy_data() -> {top_cls.__name__}: -""" - body_expr = _render_dataclass_expr(top_cls, indent=4, list_index=None) - - content = header + " return " + body_expr - - output.parent.mkdir(parents=True, exist_ok=True) - output.write_text(content, encoding="utf-8") - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Generate a Python file that builds dummy Data-like object for any mapper." - ) - parser.add_argument( - "--module-file", - required=True, - type=Path, - help="Path to the mapper module file (e.g., schema_mappers/.../mass_spectrometry.py).", - ) - parser.add_argument( - "--output", - required=True, - type=Path, - help="Path to write the generated Python file.", - ) - parser.add_argument( - "--top-class", - default="Data", - help="Top-level dataclass name to start from. Defaults to 'Data'.", - ) - args = parser.parse_args() - generate_dummy_data_py(args.module_file, args.top_class, args.output) - print(f"Wrote dummy data generator to: {args.output}") - output_path = str(args.output) - subprocess.run(["ruff", "check", "--fix", output_path], check=True) # noqa: S607 - - -if __name__ == "__main__": - main() diff --git a/scripts/generate_fake_json_from_dataclasses.py b/scripts/generate_fake_json_from_dataclasses.py deleted file mode 100644 index f659b3391..000000000 --- a/scripts/generate_fake_json_from_dataclasses.py +++ /dev/null @@ -1,248 +0,0 @@ -from __future__ import annotations - -import argparse -import dataclasses -import importlib.util -import inspect -import json -import logging -from pathlib import Path -import sys -import types -from types import ModuleType -import typing -from typing import Any, get_args, get_origin - - -def _to_space_separated(name: str) -> str: - return name.replace("_", " ").strip().lower() - - -def _load_module_from_path(module_path: Path) -> ModuleType: - spec = importlib.util.spec_from_file_location(module_path.stem, str(module_path)) - if spec is None or spec.loader is None: - msg = f"Unable to load module from {module_path}" - raise RuntimeError(msg) - module = importlib.util.module_from_spec(spec) - sys.modules[module_path.stem] = module - spec.loader.exec_module(module) - return module - - -def _is_dataclass_type(tp: Any) -> bool: - try: - return inspect.isclass(tp) and dataclasses.is_dataclass(tp) - except Exception: - return False - - -def _unwrap_optional(tp: Any) -> Any: - origin = get_origin(tp) - if origin is None: - return tp - if origin is list or origin is dict or origin is tuple: - return tp - if ( - origin is typing.Union - or str(origin) == "typing.Union" - or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) - ): - args = [a for a in get_args(tp) if a is not type(None)] - return args[0] if args else tp - return tp - - -def _choose_union_member(tp: Any) -> Any: - origin = get_origin(tp) - if origin is None: - return tp - if ( - origin is typing.Union - or str(origin) == "typing.Union" - or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) - ): - args = list(get_args(tp)) - # Prefer primitives over dataclasses for friendly fake data - for preferred in (str, int, float, bool): - if preferred in args: - return preferred - for arg in args: - if arg is not type(None): - return arg - return tp - - -def _unit_string_from_quantity_type(tp: Any) -> str | None: - # Quantity types in this codebase subclass a Units class from - # allotropy.allotrope.models.shared.definitions.units which has a default 'unit' value. - try: - for base in getattr(tp, "__mro__", []): - # Skip the quantity base itself; we need the units mixin - if base is tp: - continue - # Heuristic: a dataclass with a 'unit' field default on the class - if dataclasses.is_dataclass(base) and any( - f.name == "unit" for f in dataclasses.fields(base) - ): - # Try to read class attribute default, else instantiate - val = getattr(base, "unit", None) - if isinstance(val, str) and val: - return val - try: - instance = base() - val2 = getattr(instance, "unit", None) - if isinstance(val2, str) and val2: - return val2 - except Exception as exc: - logging.debug("Ignoring unit mixin instantiation error: %s", exc) - continue - return None - except Exception: - return None - - -def _fake_scalar_for_type( - tp: Any, field_name: str, ctx: dict[str, Any] | None = None -) -> Any: - lname = field_name.lower() - if "identifier" in lname: - base = field_name.replace("_", "-").lower() - idx = None if ctx is None else ctx.get("list_index") - if isinstance(idx, int): - return f"{base}-{idx + 1:03d}" - return f"{base}-001" - # TStringValue, TClass, TNamed → str (these are unions including str) - if tp in (str,): - # Make the fake a bit readable from the field context - return f"example {field_name.replace('_', ' ')}" - if tp in (int,): - return 1 - if tp in (float,): - return 1.0 - if tp in (bool,): - return True - return f"example {field_name.replace('_', ' ')}" - - -def _fake_for_type(tp: Any, field_name: str, ctx: dict[str, Any] | None = None) -> Any: - # Resolve Optional/Union - tp = _choose_union_member(tp) - tp = _unwrap_optional(tp) - - origin = get_origin(tp) - args = get_args(tp) - - # Lists - if origin is list and args: - inner = args[0] - items = [] - for i in range(2): - child_ctx = {} if ctx is None else dict(ctx) - child_ctx["list_index"] = i - # If the inner is a dataclass, delegate to dataclass builder to ensure nested structure - if _is_dataclass_type(_unwrap_optional(_choose_union_member(inner))): - items.append( - _build_object_for_dataclass( - _unwrap_optional(_choose_union_member(inner)), child_ctx - ) - ) - else: - items.append(_fake_for_type(inner, field_name, child_ctx)) - return items - - # Dataclasses (including nested models and quantity values) - if _is_dataclass_type(tp): - # Quantity-like dataclasses: look for 'value' and 'unit' fields (from TQuantityValue) - try: - field_names = [f.name for f in dataclasses.fields(tp)] - if "value" in field_names and "unit" in field_names: - unit = _unit_string_from_quantity_type(tp) or "unit" - return {"value": 1.0, "unit": unit} - except Exception as exc: - logging.debug("Dataclass field inspection failed: %s", exc) - # Generic nested dataclass: return its object (without wrapping in the class name) - return _build_object_for_dataclass(tp, ctx) - - # Parameterized types we don't explicitly handle: fall back - if origin is not None: - return f"example {field_name.replace('_', ' ')}" - - # Primitive or aliases that resolve to primitives - return _fake_scalar_for_type(tp, field_name, ctx) - - -def _build_object_for_dataclass( - cls: type[Any], ctx: dict[str, Any] | None = None -) -> dict[str, Any]: - result: dict[str, Any] = {} - # Resolve postponed annotations so we get real types instead of strings - try: - module_globals = sys.modules.get(cls.__module__).__dict__ - except Exception: - module_globals = None - try: - # include_extras preserves typing like list[Foo] etc. - type_hints = typing.get_type_hints( - cls, globalns=module_globals, localns=module_globals, include_extras=True - ) - except Exception: - type_hints = {} - for f in dataclasses.fields(cls): - key = _to_space_separated(f.name) - # If there is a non-None default (e.g., manifest), keep it - if f.default is not dataclasses.MISSING and f.default is not None: - result[key] = f.default - continue - annotated_type = type_hints.get(f.name, f.type) - result[key] = _fake_for_type(annotated_type, f.name, ctx) - return result - - -def generate_fake_json( - module_file: Path, top_class_name: str = "Model" -) -> dict[str, Any]: - # Ensure repo 'src' is on sys.path so imports inside the model module work - src_dir = Path(__file__).resolve().parents[1] / "src" - sys.path.insert(0, str(src_dir)) - - module = _load_module_from_path(module_file) - model_cls = getattr(module, top_class_name, None) - if model_cls is None or not _is_dataclass_type(model_cls): - msg = f"Dataclass '{top_class_name}' not found in {module_file}" - raise RuntimeError(msg) - payload = _build_object_for_dataclass(model_cls) - # Return the model's fields at the top level (no top-level 'model' key) - return payload - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Generate a fake JSON document from a dataclasses module (top class 'Model')." - ) - parser.add_argument( - "--module-file", - required=True, - type=Path, - help="Path to the Python module containing the generated dataclasses (e.g., mass_spectrometry.py)", - ) - parser.add_argument( - "--output", - required=True, - type=Path, - help="Path to write the resulting JSON file.", - ) - parser.add_argument( - "--top-class", - default="Model", - help="Top-level dataclass name to start from. Defaults to 'Model'.", - ) - args = parser.parse_args() - - data = generate_fake_json(args.module_file, args.top_class) - args.output.parent.mkdir(parents=True, exist_ok=True) - with args.output.open("w", encoding="utf-8") as f: - json.dump(data, f, indent=4, ensure_ascii=False) - - -if __name__ == "__main__": - main() diff --git a/tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py b/tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py similarity index 99% rename from tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py rename to tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py index efd92347a..4cc712e14 100644 --- a/tests/allotrope/schema_parser/mass_spectrometry_mapper_dummy_data_test.py +++ b/tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py @@ -1,5 +1,3 @@ -from __future__ import annotations - from allotropy.allotrope.allotrope import serialize_and_validate_allotrope from allotropy.allotrope.schema_mappers.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import ( Data, From 3d5c93af888167310f0758bc97e3744833db4433 Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Tue, 9 Dec 2025 13:33:36 -0500 Subject: [PATCH 4/6] Moved test file to correct path --- ...ass_spectrometry_mapper_dummy_data_test.py | 143 ------------------ 1 file changed, 143 deletions(-) delete mode 100644 tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py diff --git a/tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py b/tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py deleted file mode 100644 index 4cc712e14..000000000 --- a/tests/allotrope/schema_parser/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py +++ /dev/null @@ -1,143 +0,0 @@ -from allotropy.allotrope.allotrope import serialize_and_validate_allotrope -from allotropy.allotrope.schema_mappers.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import ( - Data, - DetectorControl, - Device, - Mapper, - Measurement, - Metadata, - Peak, - SampleIntroduction, -) -from allotropy.parsers.utils.timestamp_parser import TimestampParser - - -def build_dummy_data() -> Data: - return Data( - metadata=Metadata( - asset_management_identifier="asset-management-identifier-001", - data_processing_description="example data processing description", - devices=[ - Device( - device_type="spectrometer", - model_number="example model number", - product_manufacturer="example product manufacturer", - device_custom_info={ - "example device custom info key": "example device custom info value" - }, - ), - Device( - device_type="spectrometer", - model_number="example model number", - product_manufacturer="example product manufacturer", - device_custom_info={ - "example device custom info key": "example device custom info value" - }, - ), - ], - device_system_custom_info={ - "example device system custom info key": "example device system custom info value" - }, - ), - measurements=[ - Measurement( - analyst="example analyst", - submitter="example submitter", - measurement_mode_setting="example measurement mode setting", - sample_identifier="sample-identifier-001", - detector_control=DetectorControl( - detection_type="example detection type", - detection_duration_setting=1.0, - detector_relative_offset_setting=1.0, - detector_sampling_rate_setting=1.0, - m_z_maximum_setting=1.0, - m_z_minimum_setting=1.0, - polarity_setting="example polarity setting", - ), - sample_introduction=SampleIntroduction( - sample_introduction_medium="example sample introduction medium", - sample_introduction_mode_setting="example sample introduction mode setting", - flow_rate_setting=1.0, - laser_firing_frequency_setting=1.0, - sample_introduction_description="example sample introduction description", - ), - processed_data_types=[ - "example processed data types", - "example processed data types", - ], - data_processing_types=[ - "example data processing types", - "example data processing types", - ], - ), - Measurement( - analyst="example analyst", - submitter="example submitter", - measurement_mode_setting="example measurement mode setting", - sample_identifier="sample-identifier-002", - detector_control=DetectorControl( - detection_type="example detection type", - detection_duration_setting=1.0, - detector_relative_offset_setting=1.0, - detector_sampling_rate_setting=1.0, - m_z_maximum_setting=1.0, - m_z_minimum_setting=1.0, - polarity_setting="example polarity setting", - ), - sample_introduction=SampleIntroduction( - sample_introduction_medium="example sample introduction medium", - sample_introduction_mode_setting="example sample introduction mode setting", - flow_rate_setting=1.0, - laser_firing_frequency_setting=1.0, - sample_introduction_description="example sample introduction description", - ), - processed_data_types=[ - "example processed data types", - "example processed data types", - ], - data_processing_types=[ - "example data processing types", - "example data processing types", - ], - ), - ], - peaks=[ - Peak( - identifier="identifier-001", - m_z=1.0, - mass=1.0, - peak_area_value=1.0, - peak_area_unit="example peak area unit", - peak_height_value=1.0, - peak_height_unit="example peak height unit", - peak_width_value=1.0, - peak_width_unit="example peak width unit", - relative_peak_area=1.0, - relative_peak_height=1.0, - written_name="example written name", - ), - Peak( - identifier="identifier-002", - m_z=1.0, - mass=1.0, - peak_area_value=1.0, - peak_area_unit="example peak area unit", - peak_height_value=1.0, - peak_height_unit="example peak height unit", - peak_width_value=1.0, - peak_width_unit="example peak width unit", - relative_peak_area=1.0, - relative_peak_height=1.0, - written_name="example written name", - ), - ], - ) - - -def test_mass_spectrometry_mapper_dummy_data() -> None: - # It just validates that the mapper and serializer can handle the dummy data without raising an exception - data = build_dummy_data() - mapper_output = Mapper( - "asm_converter_name", lambda time: TimestampParser().parse(time) - ).map_model(data) - serialize_and_validate_allotrope(mapper_output) From a6b9aeb667ffb8f25529acc030bc386e1a461048 Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Wed, 10 Dec 2025 11:17:30 -0500 Subject: [PATCH 5/6] Added test file --- ...ass_spectrometry_mapper_dummy_data_test.py | 143 ++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py diff --git a/tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py b/tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py new file mode 100644 index 000000000..4cc712e14 --- /dev/null +++ b/tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py @@ -0,0 +1,143 @@ +from allotropy.allotrope.allotrope import serialize_and_validate_allotrope +from allotropy.allotrope.schema_mappers.adm.mass_spectrometry.rec._2025._06.mass_spectrometry import ( + Data, + DetectorControl, + Device, + Mapper, + Measurement, + Metadata, + Peak, + SampleIntroduction, +) +from allotropy.parsers.utils.timestamp_parser import TimestampParser + + +def build_dummy_data() -> Data: + return Data( + metadata=Metadata( + asset_management_identifier="asset-management-identifier-001", + data_processing_description="example data processing description", + devices=[ + Device( + device_type="spectrometer", + model_number="example model number", + product_manufacturer="example product manufacturer", + device_custom_info={ + "example device custom info key": "example device custom info value" + }, + ), + Device( + device_type="spectrometer", + model_number="example model number", + product_manufacturer="example product manufacturer", + device_custom_info={ + "example device custom info key": "example device custom info value" + }, + ), + ], + device_system_custom_info={ + "example device system custom info key": "example device system custom info value" + }, + ), + measurements=[ + Measurement( + analyst="example analyst", + submitter="example submitter", + measurement_mode_setting="example measurement mode setting", + sample_identifier="sample-identifier-001", + detector_control=DetectorControl( + detection_type="example detection type", + detection_duration_setting=1.0, + detector_relative_offset_setting=1.0, + detector_sampling_rate_setting=1.0, + m_z_maximum_setting=1.0, + m_z_minimum_setting=1.0, + polarity_setting="example polarity setting", + ), + sample_introduction=SampleIntroduction( + sample_introduction_medium="example sample introduction medium", + sample_introduction_mode_setting="example sample introduction mode setting", + flow_rate_setting=1.0, + laser_firing_frequency_setting=1.0, + sample_introduction_description="example sample introduction description", + ), + processed_data_types=[ + "example processed data types", + "example processed data types", + ], + data_processing_types=[ + "example data processing types", + "example data processing types", + ], + ), + Measurement( + analyst="example analyst", + submitter="example submitter", + measurement_mode_setting="example measurement mode setting", + sample_identifier="sample-identifier-002", + detector_control=DetectorControl( + detection_type="example detection type", + detection_duration_setting=1.0, + detector_relative_offset_setting=1.0, + detector_sampling_rate_setting=1.0, + m_z_maximum_setting=1.0, + m_z_minimum_setting=1.0, + polarity_setting="example polarity setting", + ), + sample_introduction=SampleIntroduction( + sample_introduction_medium="example sample introduction medium", + sample_introduction_mode_setting="example sample introduction mode setting", + flow_rate_setting=1.0, + laser_firing_frequency_setting=1.0, + sample_introduction_description="example sample introduction description", + ), + processed_data_types=[ + "example processed data types", + "example processed data types", + ], + data_processing_types=[ + "example data processing types", + "example data processing types", + ], + ), + ], + peaks=[ + Peak( + identifier="identifier-001", + m_z=1.0, + mass=1.0, + peak_area_value=1.0, + peak_area_unit="example peak area unit", + peak_height_value=1.0, + peak_height_unit="example peak height unit", + peak_width_value=1.0, + peak_width_unit="example peak width unit", + relative_peak_area=1.0, + relative_peak_height=1.0, + written_name="example written name", + ), + Peak( + identifier="identifier-002", + m_z=1.0, + mass=1.0, + peak_area_value=1.0, + peak_area_unit="example peak area unit", + peak_height_value=1.0, + peak_height_unit="example peak height unit", + peak_width_value=1.0, + peak_width_unit="example peak width unit", + relative_peak_area=1.0, + relative_peak_height=1.0, + written_name="example written name", + ), + ], + ) + + +def test_mass_spectrometry_mapper_dummy_data() -> None: + # It just validates that the mapper and serializer can handle the dummy data without raising an exception + data = build_dummy_data() + mapper_output = Mapper( + "asm_converter_name", lambda time: TimestampParser().parse(time) + ).map_model(data) + serialize_and_validate_allotrope(mapper_output) From df723ce444b1bd50b23f7356c54666056e25343c Mon Sep 17 00:00:00 2001 From: Felipe Narvaez Date: Fri, 12 Dec 2025 15:47:33 -0500 Subject: [PATCH 6/6] Moved file test again --- .../mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/allotrope/{schema_parser => schema_mappers}/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py (100%) diff --git a/tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py b/tests/allotrope/schema_mappers/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py similarity index 100% rename from tests/allotrope/schema_parser/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py rename to tests/allotrope/schema_mappers/adm/mass_spectrometry/mass_spectrometry_mapper_dummy_data_test.py