diff --git a/ord_schema/scripts/convert_udm_to_ord.py b/ord_schema/scripts/convert_udm_to_ord.py new file mode 100644 index 00000000..feede92b --- /dev/null +++ b/ord_schema/scripts/convert_udm_to_ord.py @@ -0,0 +1,884 @@ +# Copyright 2025 Open Reaction Database Project Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Builds a Dataset from a set of (UDM) Reactions. + +Given a UDM file, converts it to a Dataset .pbtxt for ORD. A UDM +may contain multiple elements; each variation is emitted as its +own ORD Reaction, since UDM variations represent distinct experiments run +under the same reaction identifiers. A may itself contain +multiple elements, which per the UDM ChangeLog model a +single dynamic, multi-stage condition profile (e.g. "20 degC for 23 min, +then 165 degC for 5 min") rather than alternative readings or separate +experiments; these are recorded via ReactionConditions.conditions_are_dynamic +and .details rather than merged into or split across a static single-step +ReactionConditions (see _add_variation). + +UDM (Unified Data Model) is a Pistoia Alliance format; see +https://github.com/PistoiaAlliance/UDM for the schema and example data. +""" + +import argparse +import getpass +import xml.etree.ElementTree as ET +from collections import defaultdict +from pathlib import Path +from typing import Any + +from ord_schema import message_helpers, updates, validations +from ord_schema.logging import get_logger +from ord_schema.proto import dataset_pb2, reaction_pb2 + +logger = get_logger(__name__) + +UdmDict = dict[str, Any] + +_MOLBLOCK_DETAILS = "MOLECULE -> MOLSTRUCTURE from UDM" + +# UDM roles that map directly onto an ORD ReactionRole. +_ROLES = { + "REACTANT": reaction_pb2.ReactionRole.REACTANT, + "REAGENT": reaction_pb2.ReactionRole.REAGENT, + "CATALYST": reaction_pb2.ReactionRole.CATALYST, + "SOLVENT": reaction_pb2.ReactionRole.SOLVENT, +} + +# UDM RXNSTRUCTURE (a child element, not an XML attribute -- see +# _build_base_reaction's structure.get("format", ...)) values that map onto +# an ORD ReactionIdentifierType. Anything not listed here (including UDM's +# default, unlabeled format) falls back to UNSPECIFIED; that reaction +# identifier is then dropped rather than emitted with an invalid type. +_RXNSTRUCTURE_FORMATS: dict[str, tuple[int, str]] = { + "cdxml": (reaction_pb2.ReactionIdentifier.CUSTOM, "cdxml"), + "rinchi": (reaction_pb2.ReactionIdentifier.RINCHI, ""), + "rsmiles": (reaction_pb2.ReactionIdentifier.REACTION_SMILES, ""), + "rxn": (reaction_pb2.ReactionIdentifier.CUSTOM, "rxn"), +} + +# Unit mappings below are taken from the UDM v6.0.0 XSD schema +# (udm_6_0_0_units.xsd, udm_6_0_0.xsd), not guessed: +# https://github.com/PistoiaAlliance/UDM/blob/master/udm_6_0_0_units.xsd + +# is UDM's molType: a decimal with an optional @unit attribute +# defaulting to "mol". Units not listed here fall back to UNSPECIFIED. +_MOLES_UNITS: dict[str, int] = { + "mol": reaction_pb2.Moles.MOLE, + "mmol": reaction_pb2.Moles.MILLIMOLE, + "umol": reaction_pb2.Moles.MICROMOLE, + "μmol": reaction_pb2.Moles.MICROMOLE, # UDM allows the literal mu character. + "nmol": reaction_pb2.Moles.NANOMOLE, +} + +# temperatureRange's @unit attribute defaults to "degC". +_TEMPERATURE_UNITS: dict[str, int] = { + "degC": reaction_pb2.Temperature.CELSIUS, + "degF": reaction_pb2.Temperature.FAHRENHEIT, + "K": reaction_pb2.Temperature.KELVIN, +} + +# pressureRange's @unit attribute defaults to "torr", not "atm". +_PRESSURE_UNITS: dict[str, int] = { + "atm": reaction_pb2.Pressure.ATMOSPHERE, + "bar": reaction_pb2.Pressure.BAR, + "psi": reaction_pb2.Pressure.PSI, + "KPa": reaction_pb2.Pressure.KILOPASCAL, + "Pa": reaction_pb2.Pressure.PASCAL, + "torr": reaction_pb2.Pressure.TORR, + "mmHg": reaction_pb2.Pressure.MM_HG, +} + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parses command-line arguments.""" + parser = argparse.ArgumentParser(description="Convert a UDM dataset to ORD") + parser.add_argument("--input", required=True, help="XML filename in UDM format") + parser.add_argument("--output", help="Output Dataset filename (*.pbtxt)") + parser.add_argument("--name", help="Name for this dataset") + parser.add_argument("--description", help="Description for this dataset") + parser.add_argument( + "--email", + help=( + "Email address of the person running this conversion. Recorded as " + "the record_created/record_modified provenance, since that person " + "-- not the original UDM scientist, who is recorded separately as " + "the experimenter -- is who is creating this ORD record." + ), + ) + parser.add_argument( + "--include-udm-xml", + action="store_true", + help=( + "If set, embed the raw UDM source XML for each reaction (and the " + "document-level UDM_VERSION/LEGAL/ORGANISATIONS/CITATIONS context " + "shared by every reaction) in provenance.reaction_metadata. Useful " + "for provenance and for humans spot-checking the converter's " + "output against the original UDM. Off by default: the source XML " + "may carry a different license than the converted dataset (see " + "the CC-BY-SA warning above), so only enable this for UDM sources " + "you have the rights to redistribute." + ), + ) + parser.add_argument( + "--no-validate", + action="store_true", + help="If set, do not run validations on reactions", + ) + return parser.parse_args(argv) + + +def main(args: argparse.Namespace) -> None: + """Converts a UDM XML dataset to an ORD Dataset and writes it to disk.""" + logger.info("Starting conversion from UDM v6.0.0 to ORD.") + logger.info( + "*** Important message - The Open Reaction Database repository, ord-data, " + "uses the CC-BY-SA license for all data. Please do not push your converted " + "dataset to ord-data unless you have the authority to change the license " + "on the data to CC-BY-SA. ***" + ) + + if not Path(args.input).is_file(): + logger.error("Conversion failed: --input file %s does not exist.", args.input) + raise SystemExit(1) + + udm_tree = ET.parse(args.input) # noqa: S314 -- UDM inputs are trusted local files, not untrusted network input. + udm_root = etree_to_dict(udm_tree.getroot()) + if "UDM" not in udm_root: + logger.error("Input file is not UDM format: expected a root element.") + raise SystemExit(1) + udm_reactions: UdmDict = udm_root["UDM"] + + if "REACTIONS" not in udm_reactions: + logger.error("No element found in input .xml file.") + raise SystemExit(1) + if "MOLECULES" not in udm_reactions: + logger.error("No element found in input .xml file.") + raise SystemExit(1) + + legal: UdmDict = udm_reactions.get("LEGAL", {}) + dataset_name = args.name or legal.get("TITLE", "") + dataset_description = args.description or ( + f"UDM dataset DOI: {legal['DOI']}" if "DOI" in legal else "" + ) + outputfile = args.output or ( + f"{legal['TITLE']}.pbtxt" if "TITLE" in legal else "ord_dataset.pbtxt" + ) + + all_molecules: dict[str, UdmDict] = {} + for molecule in _as_list(udm_reactions["MOLECULES"]["MOLECULE"]): + mol_data: UdmDict = {"name": molecule.get("NAME")} + if "MOLSTRUCTURE" in molecule: + mol_data["molblock"] = molecule["MOLSTRUCTURE"] + all_molecules[molecule["@ID"]] = mol_data + + # The person running this conversion is the one creating this ORD record, + # distinct from the original UDM scientist who performed the reaction + # (recorded separately as the experimenter). + submitter_username = getpass.getuser() + + parent_xml = ( + _document_context_xml(udm_tree.getroot()) if args.include_udm_xml else None + ) + # Parallel to _as_list(...REACTION) below: etree_to_dict and findall() both + # walk children in document order, so index i corresponds to the same + # in both. + reaction_elements: list[ET.Element] = [] + if args.include_udm_xml: + reactions_element = udm_tree.getroot().find("REACTIONS") + assert reactions_element is not None # Already validated above. + reaction_elements = reactions_element.findall("REACTION") + + pb2_reactions = [] + for i, reaction in enumerate(_as_list(udm_reactions["REACTIONS"]["REACTION"])): + reaction_xml = ( + ET.tostring(reaction_elements[i], encoding="unicode") + if args.include_udm_xml + else None + ) + base_reaction = _build_base_reaction(reaction, reaction_xml, parent_xml) + # Each variation becomes its own Reaction, seeded with the + # reaction-level identifiers built above. A reaction with no + # elements still gets a single Reaction, since it has no + # variation-specific data to add. + for variation in _as_list(reaction.get("VARIATION")) or [{}]: + pb2_reaction = reaction_pb2.Reaction() + pb2_reaction.CopyFrom(base_reaction) + _add_variation( + pb2_reaction, + reaction, + variation, + udm_reactions, + all_molecules, + submitter_username, + args.email, + ) + pb2_reactions.append(pb2_reaction) + + dataset = dataset_pb2.Dataset( + name=dataset_name, description=dataset_description, reactions=pb2_reactions + ) + # Assigns canonical dataset_id/reaction_ids and provenance updates. + updates.update_dataset(dataset) + + if not args.no_validate: + validations.validate_datasets({"_COMBINED": dataset}) + + message_helpers.save_message(dataset, outputfile) + logger.info("Conversion completed successfully: wrote %s.", outputfile) + + +def _as_list(value: Any) -> list[Any]: + """Normalizes an etree_to_dict value that may be absent, a single item, or a list. + + etree_to_dict collapses a single occurrence of a repeatable element to a + bare dict or string rather than a one-element list, so a naive `for x in + value` silently iterates a string's characters instead of treating it as + one item. Only an already-list value is left as-is. + """ + if not value: + return [] + if isinstance(value, list): + return value + return [value] + + +def _text_and_unit(value: Any) -> tuple[str | None, str | None]: + """Splits an etree_to_dict value into (text, unit). + + Several UDM elements (AMOUNT, PREPARATION) carry an optional @unit or + @format attribute alongside their text. etree_to_dict represents such an + element as plain text when no attribute is present, or as a dict with + "#text" plus "@..." keys when one is. This normalizes both shapes. + """ + if isinstance(value, dict): + return value.get("#text"), value.get("@unit") or value.get("@format") + return value, None + + +def _parse_range(value: UdmDict) -> tuple[float, float | None] | None: + """Parses a UDM min/max/exact range into a point value: (midpoint, precision). + + UDM's *Range types (temperatureRange, pressureRange, etc.) hold exactly + one of {min, max, min and max, exact} -- see Docs/ChangeLog.md's note on + XML Schema's limited support for "exactly one of" constraints. `exact` + maps to (value, None); a min/max pair maps to their midpoint with + precision = half the spread. + + A *lone* min or max is deliberately NOT treated as a point value here: + "at most 100 degC" is not the same claim as "exactly 100 degC", and + ORD's Temperature/Pressure/etc. messages have no field for a one-sided + bound, only value+precision. Returning it as (100, None) -- identical + to what an actual 100 produces -- would silently turn a + bound into a fabricated exact reading. Callers that only need a + complete/exact point value should use this function directly and treat + None as "not capturable structurally"; _format_range_text (below) + additionally renders a one-sided bound as labeled text. + """ + if "exact" in value: + return float(value["exact"]), None + min_v, max_v = value.get("min"), value.get("max") + if min_v is not None and max_v is not None: + min_f, max_f = float(min_v), float(max_v) + return (min_f + max_f) / 2, (max_f - min_f) / 2 + return None + + +def _format_range_text(value: UdmDict, default_unit: str) -> str | None: + """Renders a UDM min/max/exact range as human-readable text. + + A complete range or exact value renders as "20±5 degC" / "20 degC". A + lone min or max -- which _parse_range deliberately does not turn into a + point value, see its docstring -- renders as a labeled bound instead, + e.g. ">=20 degC" or "<=100 degC", so it's never confused with an exact + reading. + """ + unit = value.get("@unit", default_unit) + parsed = _parse_range(value) + if parsed is not None: + midpoint, precision = parsed + if precision: + return f"{midpoint}±{precision} {unit}".strip() + return f"{midpoint} {unit}".strip() + min_v, max_v = value.get("min"), value.get("max") + if min_v is not None: + return f">={min_v} {unit}".strip() + if max_v is not None: + return f"<={max_v} {unit}".strip() + return None + + +def _format_incr(value: UdmDict) -> str | None: + """Renders a *Range's optional ramp-rate child as text, if present. + + Only temperatureRange (and percentageRange, unused by CONDITION_GROUP) + define in the XSD; every other range type simply won't have it, + so this is a no-op for them. ORD has no ramp-rate field on Temperature, + so incr never has a structured home -- it must always be surfaced as + text, even when the range's central value is captured structurally. + """ + if "incr" not in value: + return None + text, unit = _text_and_unit(value["incr"]) + if not text: + return None + return f"{text} {unit}".strip() if unit else text + + +# CONDITION_GROUP's *Range-typed fields, with the default @unit each takes +# per the XSD when @unit is omitted (udm_6_0_0.xsd / udm_6_0_0_units.xsd). +_CONDITION_GROUP_RANGE_FIELDS: tuple[tuple[str, str, str], ...] = ( + ("TEMPERATURE", "degC", "temperature"), + ("PRESSURE", "torr", "pressure"), + ("TIME", "hr", "time"), + ("STIRRING", "rpm", "stirring"), + ("PH", "", "pH"), + ("REACTION_MOLARITY", "mol/L", "reaction molarity"), + ("BUFFER_CONCENTRATION", "mol/L", "buffer concentration"), + ("TOTAL_VOLUME", "L", "total volume"), +) + +# CONDITION_GROUP's Id-Ref-typed fields (references to other entities). +_CONDITION_GROUP_ID_REF_FIELDS: tuple[tuple[str, str], ...] = ( + ("REACTANT_ID", "reactants"), + ("REAGENT_ID", "reagents"), + ("CATALYST_ID", "catalysts"), + ("SOLVENT_ID", "solvents"), +) + + +def _condition_group_fields( + group: UdmDict, exclude: frozenset[str] = frozenset() +) -> list[str]: + """Renders every known CONDITION_GROUP field as human-readable text fragments. + + Covers the whole CONDITION_GROUP element (udm_6_0_0.xsd), not just the + fields ORD has a matching structured field for, so nothing is silently + dropped: `_add_conditions` (single group) and `_summarize_condition_group` + (multiple, dynamic groups) both build on this rather than each picking + their own subset. + """ + parts = [] + if "PROCESS" not in exclude and group.get("PROCESS"): + parts.append(group["PROCESS"]) + for id_field, label in _CONDITION_GROUP_ID_REF_FIELDS: + if id_field in exclude: + continue + ids = _as_list(group.get(id_field)) + if ids: + parts.append(f"{label}={','.join(ids)}") + for field, default_unit, label in _CONDITION_GROUP_RANGE_FIELDS: + field_value = group.get(field, {}) + if field not in exclude: + text = _format_range_text(field_value, default_unit) + if text: + parts.append(f"{label} {text}") + # incr (ramp rate) has no ORD equivalent regardless of whether the + # central value above is captured structurally, so it's always + # checked, not just when `field` itself isn't excluded. + incr = _format_incr(field_value) + if incr: + parts.append(f"{label} ramp={incr}") + if "REFLUX" not in exclude and "REFLUX" in group: + parts.append(f"reflux={group['REFLUX']}") + if "BUFFER_TYPE" not in exclude and group.get("BUFFER_TYPE"): + parts.append(f"buffer={group['BUFFER_TYPE']}") + if "ATMOSPHERE" not in exclude and group.get("ATMOSPHERE"): + parts.append(f"atmosphere={group['ATMOSPHERE']}") + if "PREPARATION" not in exclude: + preparations = [] + for prep in _as_list(group.get("PREPARATION")): + text, _ = _text_and_unit(prep) + if text: + preparations.append(text) + if preparations: + parts.append(f"preparation={'; '.join(preparations)}") + if "SECTION" not in exclude and "SECTION" in group: + # SECTION is UDM's generic, schema-free extension point; its content + # isn't renderable as text, but its presence is still noted. + parts.append("custom SECTION data present (see --include-udm-xml)") + return parts + + +def _summarize_condition_group(group: UdmDict, index: int) -> str: + """Renders one CONDITION_GROUP as a human-readable line for .conditions.details.""" + parts = _condition_group_fields(group) + return f"Stage {index}: " + ("; ".join(parts) if parts else "(no data)") + + +def _document_context_xml(root: ET.Element) -> str: + """Serializes the UDM document-level context shared by every reaction. + + This is everything under the root except (reaction data, + handled per-reaction) and (a lookup table already captured via + each Compound's identifiers, and often large). + """ + context = ET.Element(root.tag, root.attrib) + for child in root: + if child.tag not in ("REACTIONS", "MOLECULES"): + context.append(child) + return ET.tostring(context, encoding="unicode") + + +def _set_xml_metadata( + reaction_metadata: Any, key: str, xml_text: str, description: str +) -> None: + """Stores raw XML text as a Data entry in a reaction_metadata map.""" + data = reaction_metadata[key] + data.string_value = xml_text + data.format = "xml" + data.description = description + + +def _build_base_reaction( + reaction: UdmDict, reaction_xml: str | None, parent_xml: str | None +) -> reaction_pb2.Reaction: + """Builds the reaction-level data shared by every variation of `reaction`. + + Covers the UDM fields that live directly on rather than on a + specific : the REACTANT_ID placeholder, RXNSTRUCTURE + identifiers, and (if requested via --include-udm-xml) the raw source XML. + PRODUCT_ID is handled per-variation instead (see _add_variation), since + it needs to know which product molecules that variation's own + entries already cover, to avoid double-counting one product as two + ReactionOutcomes. + """ + base_reaction = reaction_pb2.Reaction() + + if reaction_xml: + _set_xml_metadata( + base_reaction.provenance.reaction_metadata, + "udm_reaction_xml", + reaction_xml, + "Raw UDM XML element this Reaction was converted from.", + ) + if parent_xml: + _set_xml_metadata( + base_reaction.provenance.reaction_metadata, + "udm_parent_xml", + parent_xml, + "Raw UDM document-level context (UDM_VERSION/LEGAL/ORGANISATIONS/" + "CITATIONS) shared by every reaction in the source file.", + ) + + if "REACTANT_ID" in reaction: + molinput = base_reaction.inputs["REACTANT_IDS"] + for reactant_id in _as_list(reaction["REACTANT_ID"]): + identifier = molinput.components.add().identifiers.add( + type="CUSTOM", details="REACTANT_ID from UDM" + ) + identifier.value = reactant_id + + for structure in _as_list(reaction.get("RXNSTRUCTURE")): + udm_format = structure.get("format", 1) + ordtype, orddetails = _RXNSTRUCTURE_FORMATS.get( + udm_format, (reaction_pb2.ReactionIdentifier.UNSPECIFIED, "") + ) + if ( + "value" in structure + and ordtype != reaction_pb2.ReactionIdentifier.UNSPECIFIED + ): + base_reaction.identifiers.add( + type=ordtype, details=orddetails, value=structure["value"] + ) + + return base_reaction + + +def _add_variation( + pb2_reaction: reaction_pb2.Reaction, + reaction: UdmDict, + variation: UdmDict, + udm_reactions: UdmDict, + all_molecules: dict[str, UdmDict], + submitter_username: str, + submitter_email: str | None, +) -> None: + """Populates a Reaction already seeded with base data from one VARIATION.""" + for role_name, role in _ROLES.items(): + for entry in _as_list(variation.get(role_name)): + _add_component(pb2_reaction, all_molecules, entry, role, role_name) + + conditions: UdmDict = variation.get("CONDITIONS", {}) + # is a direct child of , not . + preparations = [] + for prep in _as_list(conditions.get("PREPARATION")): + text, _ = _text_and_unit(prep) + if text: + preparations.append(text) + if preparations: + pb2_reaction.setup.environment.type = ( + reaction_pb2.ReactionSetup.ReactionEnvironment.CUSTOM + ) + pb2_reaction.setup.environment.details = "; ".join(preparations) + + # may repeat . Per Docs/ChangeLog.md's + # CONDITIONS section (Pistoia Alliance UDM repo), multiple groups model a + # single dynamic, multi-stage condition profile -- e.g. "20 degC for 23 + # min, then 165 degC for 5 min, then held 6 more min" -- not alternative + # readings or separate experiments. Merging them into one static + # ReactionConditions or splitting each into its own Reaction would both + # fabricate data that isn't in the source. ORD models exactly this case + # with conditions_are_dynamic + a free-text details field ("e.g., + # multiple stages" per its own doc comment), so the full staged profile + # is recorded as text instead. + condition_groups = _as_list(conditions.get("CONDITION_GROUP")) + if len(condition_groups) == 1: + captured = _add_conditions(pb2_reaction, condition_groups[0]) + # Whatever _add_conditions didn't actually manage to capture + # structurally -- either because ORD has no field for it (e.g. + # BUFFER_TYPE) or because this particular value couldn't be, despite + # the field type usually being structural (e.g. STIRRING in a unit + # other than rpm) -- still gets recorded as text, not dropped. + extra = _condition_group_fields(condition_groups[0], exclude=captured) + if extra: + pb2_reaction.conditions.details = "; ".join(extra) + elif len(condition_groups) > 1: + pb2_reaction.conditions.conditions_are_dynamic = True + pb2_reaction.conditions.details = "; ".join( + _summarize_condition_group(group, i) + for i, group in enumerate(condition_groups, start=1) + ) + + comment = variation.get("COMMENT") + if comment: + pb2_reaction.observations.add().comment = comment + + # PRODUCT_ID is a reaction-level placeholder (bare molecule ID, no + # structure/yield); PRODUCT is the fuller variation-level data. Track + # which molecules this variation's own PRODUCT entries already cover so + # a PRODUCT_ID for the same molecule isn't recorded as a second, + # duplicate ReactionOutcome. + covered_product_ids = set() + for udm_product in _as_list(variation.get("PRODUCT")): + molecule = udm_product.get("MOLECULE") + if molecule and molecule.get("@MOL_ID"): + covered_product_ids.add(molecule["@MOL_ID"]) + _add_product(pb2_reaction, all_molecules, udm_product) + + for product_id in _as_list(reaction.get("PRODUCT_ID")): + if product_id in covered_product_ids: + continue + identifier = ( + pb2_reaction.outcomes.add() + .products.add() + .identifiers.add(type="CUSTOM", details="PRODUCT_ID from UDM") + ) + identifier.value = product_id + + _add_provenance( + pb2_reaction, + reaction, + variation, + udm_reactions, + submitter_username, + submitter_email, + ) + pb2_reaction.provenance.is_mined = False + + +def _add_component( + pb2_reaction: reaction_pb2.Reaction, + all_molecules: dict[str, UdmDict], + entry: UdmDict, + role: int, + role_label: str, +) -> None: + """Adds one UDM REACTANT/REAGENT/CATALYST/SOLVENT entry as a ReactionInput.""" + if "MOLECULE" not in entry: + return + mol_id = entry["MOLECULE"]["@MOL_ID"] + # ReactionInput map keys are just descriptions (see reaction.proto), but + # components sharing one ReactionInput are modeled as added together in + # one addition event. If this MOL_ID already has a component under a + # *different* role, reuse would misrepresent two separate additions + # (e.g. the same compound used as both a REAGENT and a CATALYST, with + # distinct amounts) as one; disambiguate the key in that case only, so + # the common (non-colliding) case keeps the plain MOL_ID key. + key = mol_id + existing = pb2_reaction.inputs.get(mol_id) + if existing and any(c.reaction_role != role for c in existing.components): + key = f"{mol_id}_{role_label}" + component = pb2_reaction.inputs[key].components.add() + _set_molecule_identifier(component, all_molecules, mol_id, role_label) + component.reaction_role = role + if "AMOUNT" in entry: + # UDM's (molType) is molar, defaulting to "mol"; it is not a + # mass. See _MOLES_UNITS. + text, unit = _text_and_unit(entry["AMOUNT"]) + if text: + component.amount.moles.value = float(text) + component.amount.moles.units = _MOLES_UNITS.get( + unit or "mol", reaction_pb2.Moles.UNSPECIFIED + ) + + +def _set_molecule_identifier( + component: reaction_pb2.Compound | reaction_pb2.ProductCompound, + all_molecules: dict[str, UdmDict], + mol_id: str, + role_label: str, +) -> None: + """Sets a CompoundIdentifier on `component`, falling back if `mol_id` is unknown.""" + molval = all_molecules.get(mol_id) + if molval and molval.get("molblock"): + identifier = component.identifiers.add( + type="MOLBLOCK", details=_MOLBLOCK_DETAILS + ) + identifier.value = molval["molblock"] + elif molval and molval.get("name"): + identifier = component.identifiers.add( + type="CUSTOM", details=f"{role_label} MOLECULE NAME from UDM" + ) + identifier.value = molval["name"] + elif molval is not None: + # The molecule exists in but has neither a name nor a + # structure -- distinct from mol_id not being in at all. + identifier = component.identifiers.add( + type="CUSTOM", + details=( + f"{role_label} MOL_ID from UDM (found in but has " + "no NAME or MOLSTRUCTURE)" + ), + ) + identifier.value = mol_id + else: + identifier = component.identifiers.add( + type="CUSTOM", + details=f"{role_label} MOL_ID from UDM (not found in )", + ) + identifier.value = mol_id + + +def _add_conditions( + pb2_reaction: reaction_pb2.Reaction, condition_group: UdmDict +) -> frozenset[str]: + """Populates ReactionConditions from a UDM CONDITION_GROUP. + + Returns the CONDITION_GROUP field names actually captured structurally, + so the caller knows what's safe to omit from a free-text fallback (and, + just as importantly, what wasn't captured despite being present and + still needs to be surfaced as text -- e.g. STIRRING in a unit other than + rpm, which ORD's StirringConditions.rate.rpm can't represent). + """ + captured = set() + + temperature = condition_group.get("TEMPERATURE", {}) + parsed = _parse_range(temperature) + if parsed: + value, precision = parsed + pb2_reaction.conditions.temperature.setpoint.value = value + if precision: + pb2_reaction.conditions.temperature.setpoint.precision = precision + pb2_reaction.conditions.temperature.setpoint.units = _TEMPERATURE_UNITS.get( + temperature.get("@unit", "degC"), reaction_pb2.Temperature.UNSPECIFIED + ) + captured.add("TEMPERATURE") + + pressure = condition_group.get("PRESSURE", {}) + parsed = _parse_range(pressure) + if parsed: + value, precision = parsed + pb2_reaction.conditions.pressure.setpoint.value = value + if precision: + pb2_reaction.conditions.pressure.setpoint.precision = precision + pb2_reaction.conditions.pressure.setpoint.units = _PRESSURE_UNITS.get( + pressure.get("@unit", "torr"), reaction_pb2.Pressure.UNSPECIFIED + ) + captured.add("PRESSURE") + + # UDM's STIRRING is a numeric rate (stirringRange, @unit defaulting to + # "rpm"), not free text; ORD's StirringConditions.rate.rpm is the closest + # match, but only when the unit actually is rpm -- stirringRange's @unit + # is unrestricted xs:string in the XSD, not an enum, so other units are + # possible and can't be represented structurally. + stirring = condition_group.get("STIRRING", {}) + parsed = _parse_range(stirring) + if parsed and stirring.get("@unit", "rpm") == "rpm": + value, precision = parsed + rpm = round(value) + pb2_reaction.conditions.stirring.type = reaction_pb2.StirringConditions.CUSTOM + pb2_reaction.conditions.stirring.details = f"{rpm} rpm" + pb2_reaction.conditions.stirring.rate.rpm = rpm + # rate.rpm is a bare int with no precision field, unlike + # Temperature/Pressure's setpoint -- a real min/max spread (e.g. + # 490-510) would be silently rounded away to 500 with no trace + # unless STIRRING is left out of `captured` here, letting it fall + # through to the text fallback, which does render the full range. + if not precision: + captured.add("STIRRING") + + if "REFLUX" in condition_group: + pb2_reaction.conditions.reflux = _parse_bool(condition_group["REFLUX"]) + captured.add("REFLUX") + + ph = condition_group.get("PH", {}) + parsed = _parse_range(ph) + if parsed: + pb2_reaction.conditions.ph = parsed[0] + captured.add("PH") + + return frozenset(captured) + + +def _parse_bool(value: str) -> bool: + """Parses a UDM boolean-ish string (e.g. "true"/"false"/"1"/"0").""" + return str(value).strip().lower() in ("true", "1", "yes") + + +def _add_product( + pb2_reaction: reaction_pb2.Reaction, + all_molecules: dict[str, UdmDict], + udm_product: UdmDict, +) -> None: + """Adds one UDM VARIATION/PRODUCT entry as a ReactionOutcome product.""" + molecule = udm_product.get("MOLECULE") + if not molecule: + return + component = pb2_reaction.outcomes.add().products.add() + _set_molecule_identifier( + component, all_molecules, molecule.get("@MOL_ID"), "PRODUCT" + ) + yield_ = udm_product.get("YIELD", {}) + parsed = _parse_range(yield_) + if parsed: + value, precision = parsed + measurement = component.measurements.add() + measurement.type = reaction_pb2.ProductMeasurement.ProductMeasurementType.YIELD + # ORD expects YIELD as a percentage (validations.py warns otherwise); + # UDM's YIELD is already percentage-scale. + measurement.percentage.value = value + if precision: + measurement.percentage.precision = precision + else: + # A lone or ("at least"/"at most" X%) isn't a point + # value _parse_range will produce (see its docstring) -- ORD's + # Percentage has no field for a one-sided bound, so it's recorded + # as text via .details instead of being fabricated as exact or + # silently dropped. + text = _format_range_text(yield_, "percent") + if text: + measurement = component.measurements.add() + measurement.type = ( + reaction_pb2.ProductMeasurement.ProductMeasurementType.YIELD + ) + measurement.details = f"yield {text}" + + +def _add_provenance( + pb2_reaction: reaction_pb2.Reaction, + reaction: UdmDict, + variation: UdmDict, + udm_reactions: UdmDict, + submitter_username: str, + submitter_email: str | None, +) -> None: + """Populates ReactionProvenance from reaction- and variation-level UDM fields.""" + # record_created/record_modified describe who is creating this ORD + # record -- i.e. whoever is running this conversion -- not the original + # UDM scientist, who is recorded separately below as the experimenter. + pb2_reaction.provenance.record_created.person.username = submitter_username + if submitter_email: + pb2_reaction.provenance.record_created.person.email = submitter_email + + legal: UdmDict = udm_reactions.get("LEGAL", {}) + producer = legal.get("PRODUCER") + if producer: + # Fallback organization; overridden below if the scientist has their + # own affiliation, since PRODUCER is the data vendor (e.g. "REAXYS"), + # not necessarily the scientist's research institution. + pb2_reaction.provenance.experimenter.organization = producer + + # SCIENTIST is UDM's authorDetails type (NAME/EMAIL/PHONE/ORGANISATION), + # not a plain string, and may repeat; the first is taken as the + # experimenter of record. + scientists = _as_list(variation.get("SCIENTIST")) + scientist = scientists[0] if scientists else None + if isinstance(scientist, dict): + if scientist.get("NAME"): + pb2_reaction.provenance.experimenter.name = scientist["NAME"] + if scientist.get("EMAIL"): + pb2_reaction.provenance.experimenter.email = scientist["EMAIL"] + organisation = scientist.get("ORGANISATION") + if isinstance(organisation, dict): + if organisation.get("NAME"): + pb2_reaction.provenance.experimenter.organization = organisation["NAME"] + if organisation.get("ADDRESS"): + pb2_reaction.provenance.city = organisation["ADDRESS"] + elif scientist: + # Defensive fallback in case a non-conformant UDM file has a bare + # string SCIENTIST rather than the schema's authorDetails structure. + pb2_reaction.provenance.experimenter.name = scientist + + doi = legal.get("DOI") + if doi and "CITATIONS" not in reaction: + pb2_reaction.provenance.doi = doi + elif "CITATIONS" in udm_reactions and "CITATION" in variation: + variation_citation = variation["CITATION"] + if isinstance(variation_citation, list): + variation_citation = variation_citation[0] + for citation in _as_list(udm_reactions["CITATIONS"]["CITATION"]): + if ( + citation.get("@ID") == variation_citation.get("@CIT_ID") + and "DOI" in citation + ): + pb2_reaction.provenance.doi = citation["DOI"] + break + + # reaction.get("CITATIONS") may be None (empty ); its + # "CITATION" child, not CITATIONS itself, is the repeatable element. + reaction_citations = _as_list((reaction.get("CITATIONS") or {}).get("CITATION")) + if reaction_citations: + reaction_citation = reaction_citations[0] + if "DOI" in reaction_citation: + pb2_reaction.provenance.doi = reaction_citation["DOI"] + if "PATENT_NUMBER" in reaction_citation: + pb2_reaction.provenance.patent = reaction_citation["PATENT_NUMBER"] + + creation_date = variation.get("CREATION_DATE") + if creation_date: + pb2_reaction.provenance.record_created.time.value = creation_date + + modification_date = variation.get("MODIFICATION_DATE") + if modification_date: + event = pb2_reaction.provenance.record_modified.add() + event.time.value = modification_date + event.person.username = submitter_username + if submitter_email: + event.person.email = submitter_email + + +# Converts XML tree into Python dictionary. +# https://stackoverflow.com/questions/7684333/converting-xml-to-dictionary-using-elementtree +def etree_to_dict(t: ET.Element) -> UdmDict: + """Recursively converts an ElementTree Element into a nested dict.""" + d: UdmDict = {t.tag: {} if t.attrib else None} + children = list(t) + if children: + dd = defaultdict(list) + for dc in map(etree_to_dict, children): + for k, v in dc.items(): + dd[k].append(v) + d = {t.tag: {k: v[0] if len(v) == 1 else v for k, v in dd.items()}} + if t.attrib: + d[t.tag].update(("@" + k, v) for k, v in t.attrib.items()) + if t.text: + text = t.text.strip() + if children or t.attrib: + if text: + d[t.tag]["#text"] = text + else: + d[t.tag] = text + return d + + +if __name__ == "__main__": + main(parse_args()) diff --git a/ord_schema/scripts/convert_udm_to_ord_test.py b/ord_schema/scripts/convert_udm_to_ord_test.py new file mode 100644 index 00000000..103d1823 --- /dev/null +++ b/ord_schema/scripts/convert_udm_to_ord_test.py @@ -0,0 +1,1025 @@ +# Copyright 2025 Open Reaction Database Project Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for ord_schema.scripts.convert_udm_to_ord.""" + +import getpass +import pathlib + +import pytest + +from ord_schema import message_helpers +from ord_schema.proto import dataset_pb2, reaction_pb2 +from ord_schema.scripts import convert_udm_to_ord + +_MOLECULES = """ + + Reactant A + Product A + +""" + +# SCIENTIST is UDM's authorDetails type (NAME/EMAIL/PHONE/ORGANISATION), not a +# bare string -- see udm_6_0_0.xsd. +_SCIENTIST = """ + + Test Scientist + scientist@example.com + + Scientist Org +
123 Lab St
+ + +""" + +_ONE_VARIATION_REACTION = f""" + + + {_SCIENTIST} + 2020-01-01 + + Reactant A + 1.0 + + + Product A + 85.0 + + + Stirred under nitrogen. + + 298.0 + 1.5 + 500 + + + A test reaction + + +""" + + +def _udm_xml(reactions_xml: str) -> str: + """Wraps a ... fragment in a minimal UDM document.""" + return f""" + + Test Dataset + Data Vendor + 10.0000/test-doi + + {_MOLECULES} + + {reactions_xml} + + +""" + + +def _write(tmp_path: pathlib.Path, filename: str, content: str) -> str: + path = tmp_path / filename + path.write_text(content) + return str(path) + + +def test_simple(tmp_path): + input_filename = _write(tmp_path, "input.xml", _udm_xml(_ONE_VARIATION_REACTION)) + output_filename = str(tmp_path / "output.pbtxt") + argv = [ + "--input", + input_filename, + "--output", + output_filename, + "--email", + "submitter@example.com", + ] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert dataset.name == "Test Dataset" + assert dataset.description == "UDM dataset DOI: 10.0000/test-doi" + assert len(dataset.reactions) == 1 + + reaction = dataset.reactions[0] + # The experimenter is the original UDM scientist who ran the reaction, + # with their own organization/email/address taking precedence over the + # dataset-level PRODUCER (a data vendor, not a research institution). + assert reaction.provenance.experimenter.name == "Test Scientist" + assert reaction.provenance.experimenter.email == "scientist@example.com" + assert reaction.provenance.experimenter.organization == "Scientist Org" + assert reaction.provenance.city == "123 Lab St" + + # record_created is whoever ran this conversion, not the UDM scientist. + assert reaction.provenance.record_created.person.username == getpass.getuser() + assert reaction.provenance.record_created.person.email == "submitter@example.com" + assert reaction.provenance.record_created.person.name != "Test Scientist" + assert reaction.provenance.record_created.time.value == "2020-01-01" + + assert reaction.conditions.temperature.setpoint.value == 298.0 + assert ( + reaction.conditions.temperature.setpoint.units + == reaction.conditions.temperature.setpoint.KELVIN + ) + assert reaction.conditions.pressure.setpoint.value == 1.5 + assert ( + reaction.conditions.pressure.setpoint.units + == reaction.conditions.pressure.setpoint.BAR + ) + assert reaction.conditions.stirring.rate.rpm == 500.0 + assert reaction.setup.environment.details == "Stirred under nitrogen." + assert reaction.observations[0].comment == "A test reaction" + + [component] = reaction.inputs["m1"].components + assert component.amount.moles.value == 1.0 + assert component.amount.moles.units == component.amount.moles.MOLE + + [outcome] = reaction.outcomes + [product] = outcome.products + assert product.identifiers[0].value == "Product A" + [measurement] = product.measurements + assert measurement.percentage.value == 85.0 + + # Off by default: no raw UDM XML embedded without --include-udm-xml. + assert "udm_reaction_xml" not in reaction.provenance.reaction_metadata + assert "udm_parent_xml" not in reaction.provenance.reaction_metadata + + +def test_include_udm_xml(tmp_path): + reaction_xml = """ + + + Scientist One + + Reactant A + + + + Scientist Two + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = [ + "--input", + input_filename, + "--output", + output_filename, + "--no-validate", + "--include-udm-xml", + ] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions) == 2 + + for reaction in dataset.reactions: + metadata = reaction.provenance.reaction_metadata + assert metadata["udm_reaction_xml"].format == "xml" + assert "" in metadata["udm_reaction_xml"].string_value + # Both VARIATIONs came from the same , so they share + # identical reaction-level XML, including both scientists. + assert "Scientist One" in metadata["udm_reaction_xml"].string_value + assert "Scientist Two" in metadata["udm_reaction_xml"].string_value + assert metadata["udm_parent_xml"].format == "xml" + assert "Test Dataset" in metadata["udm_parent_xml"].string_value + assert "Data Vendor" in metadata["udm_parent_xml"].string_value + # MOLECULES is deliberately excluded: it's a lookup table, not + # reaction-specific context, and can be large. + assert "MOLECULES" not in metadata["udm_parent_xml"].string_value + + # The two reactions' udm_reaction_xml differ only by REACTION-vs-VARIATION + # scope, not by content -- both variations share one element. + reaction_xmls = { + r.provenance.reaction_metadata["udm_reaction_xml"].string_value + for r in dataset.reactions + } + assert len(reaction_xmls) == 1 + + +def test_scientist_organization_falls_back_to_producer(tmp_path): + """When the scientist has no inline ORGANISATION, PRODUCER fills in.""" + reaction_xml = """ + + + No Org Scientist + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + assert reaction.provenance.experimenter.name == "No Org Scientist" + assert reaction.provenance.experimenter.organization == "Data Vendor" + + +def test_record_created_defaults_to_submitter_without_email(tmp_path): + """Without --email, record_created still identifies the submitter (by OS + username) rather than the UDM scientist; the dataset just won't pass + full validation without an explicit --email, since record_created still + needs one even though the scientist's own email (if UDM supplies it) + goes to experimenter, not record_created. + """ + input_filename = _write(tmp_path, "input.xml", _udm_xml(_ONE_VARIATION_REACTION)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + assert reaction.provenance.record_created.person.username == getpass.getuser() + assert not reaction.provenance.record_created.person.email + assert reaction.provenance.experimenter.name == "Test Scientist" + assert reaction.provenance.experimenter.email == "scientist@example.com" + + +def test_multiple_variations_become_separate_reactions(tmp_path): + """Regression test: each must produce its own Reaction. + + A prior version of this script reset the in-progress Reaction object on + every VARIATION and only kept the last one, silently discarding every + other variation's data. + """ + reaction_xml = """ + + + Scientist One + + Reactant A + 1.0 + + + Product A + + + + Scientist Two + + Reactant A + 2.0 + + + Product A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions) == 2 + scientists = {r.provenance.experimenter.name for r in dataset.reactions} + assert scientists == {"Scientist One", "Scientist Two"} + amounts = { + r.inputs["m1"].components[0].amount.moles.value for r in dataset.reactions + } + assert amounts == {1.0, 2.0} + + +def test_multiple_condition_groups_are_recorded_as_dynamic(tmp_path): + """Regression test: multiple s model one dynamic, + multi-stage profile -- see Docs/ChangeLog.md's CONDITIONS section in the + UDM repo -- not alternative readings or separate experiments. + + Two earlier versions of this code got this wrong in opposite ways: one + merged every group's fields into one static ReactionConditions + (fabricating composites, e.g. one group's temperature plus an unrelated + group's pressure, that never existed in the source); another split each + group into its own Reaction, which duplicated the variation-level + REACTANT/PRODUCT/YIELD/COMMENT/provenance into every split, asserting a + single recorded outcome was independently reproduced under each + condition set. Neither merges nor splits: conditions_are_dynamic and a + full-fidelity text summary in .details record the staged profile, per + the fields ORD itself provides for exactly this case. + """ + reaction_xml = """ + + + Test Scientist + + Reactant A + + + Product A + 85.0 + + + + 2020 + + + + 165165 + + + + + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions) == 1 + reaction = dataset.reactions[0] + + # Only the outcome recorded once in the source, not duplicated. + assert len(reaction.outcomes) == 1 + assert reaction.outcomes[0].products[0].measurements[0].percentage.value == 85.0 + + assert reaction.conditions.conditions_are_dynamic + # No single-step snapshot: none of the three stages is "the" temperature. + assert reaction.conditions.temperature.setpoint.value == 0.0 + assert "Stage 1: temperature 20.0 degC; time 23.0 hr" in reaction.conditions.details + assert "Stage 2: temperature 165.0 degC; time 5.0 hr" in reaction.conditions.details + assert "Stage 3: time 6.0 hr" in reaction.conditions.details + + +def test_condition_group_min_max_range_uses_midpoint_and_precision(tmp_path): + """A single with a min/max range (no ) -- the + form UDM's own documentation example uses -- must still populate the + structured setpoint, not just . + """ + reaction_xml = """ + + + + Reactant A + + + + 2030 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + assert not reaction.conditions.conditions_are_dynamic + assert reaction.conditions.temperature.setpoint.value == 25.0 + assert reaction.conditions.temperature.setpoint.precision == 5.0 + + +def test_condition_group_buffer_and_agent_fields_are_not_dropped(tmp_path): + """Regression test: CONDITION_GROUP fields ORD has no structured field for + (BUFFER_TYPE, BUFFER_CONCENTRATION, REACTION_MOLARITY, TOTAL_VOLUME, + REAGENT_ID/...) must still show up somewhere, not vanish silently just + because they aren't temperature/pressure/stirring/reflux/pH. + """ + reaction_xml = """ + + + + Reactant A + + + + 25.0 + r1 + phosphate + 0.1 + 0.5 + 10 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # Still structurally represented, same as before. + assert reaction.conditions.temperature.setpoint.value == 25.0 + # Everything else lands in details rather than disappearing. + details = reaction.conditions.details + assert "reagents=r1" in details + assert "buffer=phosphate" in details + assert "buffer concentration 0.1 mol/L" in details + assert "reaction molarity 0.5 mol/L" in details + assert "total volume 10.0 L" in details + # Not duplicated: temperature is structural only, not also in details. + assert "temperature" not in details + + +def test_dynamic_condition_groups_include_buffer_and_agent_fields(tmp_path): + """Same as above, but for the multi-group (conditions_are_dynamic) path, + where every field -- including temperature/pressure/stirring/pH -- has + to come from the text summary, since nothing is structurally set there. + """ + reaction_xml = """ + + + + Reactant A + + + + 20.0 + phosphate + + + 40.0 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + assert reaction.conditions.conditions_are_dynamic + assert "buffer=phosphate" in reaction.conditions.details + + +def test_temperature_ramp_incr_is_not_dropped(tmp_path): + """Regression test: 's optional ramp-rate child has + no ORD field to live in, so it must surface as text even in the + single-group case where the central value *is* captured structurally + (Temperature.setpoint) -- excluding all of TEMPERATURE from the "extra" + text pass would silently drop incr along with it. + + Uses a complete / pair rather than , since the XSD's + temperatureRange only allows alongside min/max, never . + """ + reaction_xml = """ + + + + Reactant A + + + + + 95 + 105 + 5 + + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # Central value still structural: a complete min/max pair is a point + # value (midpoint + precision), unlike a lone bound (see the next test). + assert reaction.conditions.temperature.setpoint.value == 100.0 + assert reaction.conditions.temperature.setpoint.precision == 5.0 + # Ramp rate, which has no structured home, surfaces as text instead. + assert "temperature ramp=5 deg_C/hour" in reaction.conditions.details + + +def test_lone_bound_is_not_treated_as_an_exact_value(tmp_path): + """Regression test: a lone (no , no ) means "at most + X", not "exactly X" -- it must not be written to Temperature.setpoint + as if it were an exact reading, and must still be recoverable as + labeled text rather than silently dropped. + """ + reaction_xml = """ + + + + Reactant A + + + + 100 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # Not fabricated as an exact 100 degC setpoint. + assert reaction.conditions.temperature.setpoint.value == 0.0 + # But not silently dropped either -- recorded as a labeled bound. + assert "temperature <=100 degC" in reaction.conditions.details + + +def test_lone_bound_in_dynamic_stage_is_labeled(tmp_path): + """Same lone-bound distinction, but in the multi-group (dynamic) text + summary, which is the only record of each stage. + """ + reaction_xml = """ + + + + Reactant A + + + + 2020 + + + 90 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + details = dataset.reactions[0].conditions.details + assert "Stage 1: temperature 20.0 degC" in details + assert "Stage 2: temperature >=90 degC" in details + + +def test_non_rpm_stirring_is_not_dropped(tmp_path): + """Regression test: STIRRING in a unit other than rpm has no structured + home (StirringConditions.rate.rpm is rpm-specific), so it must surface + as text -- unlike rpm-unit stirring, which _add_conditions does capture + structurally, this one isn't actually captured despite STIRRING usually + being a "structural" field, so it can't be blanket-excluded from the + text pass the way BUFFER_TYPE etc. are (see _add_conditions' return + value, which _add_variation checks rather than assuming). + """ + reaction_xml = """ + + + + Reactant A + + + + 50 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # Not structurally represented: rate.rpm can't hold a Hz value. + assert reaction.conditions.stirring.rate.rpm == 0 + # But not silently dropped either. + assert "stirring 50.0 Hz" in reaction.conditions.details + + +def test_stirring_range_precision_is_not_dropped(tmp_path): + """Regression test: an rpm STIRRING given as a / range must not + be silently rounded down to a bare exact value -- StirringConditions. + rate.rpm has no precision field (unlike Temperature/Pressure's + setpoint), so the range would otherwise vanish, leaving 490-510 rpm + indistinguishable from an exact 500 rpm. + """ + reaction_xml = """ + + + + Reactant A + + + + 490510 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # Best-effort structural value: the rounded midpoint. + assert reaction.conditions.stirring.rate.rpm == 500 + # But the range itself isn't lost -- it's still in the text fallback, + # as midpoint and precision (avoid asserting the literal "±" glyph). + assert "stirring 500.0" in reaction.conditions.details + assert "10.0 rpm" in reaction.conditions.details + + +def test_stirring_exact_value_is_not_duplicated_as_text(tmp_path): + """An exact (or degenerate min==max) rpm STIRRING loses nothing by being + rounded, so it should be captured structurally only, not also repeated + in the text fallback. + """ + reaction_xml = """ + + + + Reactant A + + + + 500 + + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + assert reaction.conditions.stirring.rate.rpm == 500 + assert "stirring" not in reaction.conditions.details + + +def test_reaction_without_variation_still_emits_one_reaction(tmp_path): + reaction_xml = """ + + + rsmiles + CC.O>>CCO + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions) == 1 + assert dataset.reactions[0].identifiers[0].value == "CC.O>>CCO" + + +def test_single_reactant_id_is_not_split_into_characters(tmp_path): + """Regression test: a lone must not be iterated as a string. + + UDM's REACTANT_ID/PRODUCT_ID are maxOccurs="unbounded"; etree_to_dict + collapses a single occurrence to a bare string rather than a one-element + list, so naive iteration over it would previously produce one bogus + identifier per character. + """ + reaction_xml = """ + + abc123 + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [component] = dataset.reactions[0].inputs["REACTANT_IDS"].components + assert len(component.identifiers) == 1 + assert component.identifiers[0].value == "abc123" + + +def test_empty_product_id_does_not_add_spurious_outcome(tmp_path): + """Regression test: a self-closing must not add an empty + ReactionOutcome with zero products. + """ + reaction_xml = """ + + + + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions[0].outcomes) == 0 + + +def test_product_id_does_not_duplicate_a_covered_product(tmp_path): + """Regression test: a reaction-level PRODUCT_ID for a molecule already + covered by the variation's own must not add a second, + duplicate ReactionOutcome for the same product. + """ + reaction_xml = """ + + m2 + + + Reactant A + + + Product A + 85.0 + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [outcome] = dataset.reactions[0].outcomes + [product] = outcome.products + assert product.measurements[0].percentage.value == 85.0 + + +def test_product_id_still_recorded_when_not_covered_by_product(tmp_path): + """A PRODUCT_ID for a molecule the variation's own PRODUCT list doesn't + mention is still recorded as its own placeholder outcome. + """ + reaction_xml = """ + + m2 + + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [outcome] = dataset.reactions[0].outcomes + assert outcome.products[0].identifiers[0].value == "m2" + + +def test_unknown_molecule_reference_does_not_crash(tmp_path): + reaction_xml = """ + + + + Missing + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [component] = dataset.reactions[0].inputs["does_not_exist"].components + assert component.identifiers[0].type == component.identifiers[0].CUSTOM + assert component.identifiers[0].value == "does_not_exist" + + +def test_molecule_found_but_empty_is_not_mislabeled_as_missing(tmp_path): + """Regression test: a present in but with neither + a NAME nor a MOLSTRUCTURE must not be labeled "not found in " + -- it was found, it's just empty, which is a different, distinguishable + problem for anyone debugging conversion gaps. + """ + molecules = """ + + + + """ + reaction_xml = """ + + + + Reactant A + + + +""" + udm_xml = f""" + Test Dataset + {molecules} + {reaction_xml} + +""" + input_filename = _write(tmp_path, "input.xml", udm_xml) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [component] = dataset.reactions[0].inputs["m1"].components + details = component.identifiers[0].details + assert "not found in " not in details + assert "found in but has no NAME or MOLSTRUCTURE" in details + + +def test_same_molecule_in_two_roles_does_not_merge_inputs(tmp_path): + """Regression test: the same MOL_ID used under two different roles in one + variation (e.g. as both a REAGENT and a CATALYST) must not collapse into + a single ReactionInput, which would misrepresent two separate additions + (with distinct amounts) as one addition event. + """ + reaction_xml = """ + + + + Reactant A + 1.0 + + + Reactant A + 0.05 + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + reaction = dataset.reactions[0] + # The first-seen role (REAGENT) keeps the plain MOL_ID key; the second, + # different role (CATALYST) gets a disambiguated key instead of merging + # into the same ReactionInput. + assert len(reaction.inputs) == 2 + [reagent_component] = reaction.inputs["m1"].components + assert reagent_component.reaction_role == reaction_pb2.ReactionRole.REAGENT + assert reagent_component.amount.moles.value == 1.0 + + [other_key] = [key for key in reaction.inputs if key != "m1"] + assert other_key == "m1_CATALYST" + [catalyst_component] = reaction.inputs[other_key].components + assert catalyst_component.reaction_role == reaction_pb2.ReactionRole.CATALYST + assert catalyst_component.amount.moles.value == pytest.approx(0.05) + + +def test_yield_min_max_range_is_not_dropped(tmp_path): + """Regression test: a YIELD given as / (no ) must still + produce a measurement, using percentage (not float_value) so it's + readable via message_helpers.get_product_yield(), the standard accessor. + """ + reaction_xml = """ + + + + Product A + 8090 + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [measurement] = dataset.reactions[0].outcomes[0].products[0].measurements + assert measurement.percentage.value == 85.0 + assert measurement.percentage.precision == 5.0 + assert ( + message_helpers.get_product_yield(dataset.reactions[0].outcomes[0].products[0]) + == 85.0 + ) + + +def test_yield_lone_bound_is_not_fabricated_as_exact(tmp_path): + """Regression test: a YIELD given as a lone ("at least X%") must + not be written to Percentage.value as if it were an exact reading -- + Percentage has no field for a one-sided bound, so it's recorded as text + via ProductMeasurement.details instead. + """ + reaction_xml = """ + + + + Product A + 80 + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + [measurement] = dataset.reactions[0].outcomes[0].products[0].measurements + assert not measurement.HasField("percentage") + assert measurement.details == "yield >=80 percent" + + +def test_reaction_level_citations_with_multiple_entries(tmp_path): + """Regression test: a reaction-level with more than one + child must still resolve a DOI/patent from the first one, + not silently drop them because the wrong dict level got _as_list'd. + """ + reaction_xml = """ + + + 10.0000/firstUS123 + 10.0000/second + + + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert dataset.reactions[0].provenance.doi == "10.0000/first" + assert dataset.reactions[0].provenance.patent == "US123" + + +def test_reaction_level_empty_citations_does_not_crash(tmp_path): + """Regression test: a self-closing must not crash the + conversion with an IndexError. + """ + reaction_xml = """ + + + + + Reactant A + + + +""" + input_filename = _write(tmp_path, "input.xml", _udm_xml(reaction_xml)) + output_filename = str(tmp_path / "output.pbtxt") + argv = ["--input", input_filename, "--output", output_filename, "--no-validate"] + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + dataset = message_helpers.load_message(output_filename, dataset_pb2.Dataset) + assert len(dataset.reactions) == 1 + assert not dataset.reactions[0].provenance.doi + + +def test_missing_input_file(tmp_path): + argv = ["--input", str(tmp_path / "does_not_exist.xml")] + with pytest.raises(SystemExit): + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv)) + + +def test_not_udm_format(tmp_path): + input_filename = _write(tmp_path, "input.xml", "") + argv = ["--input", input_filename] + with pytest.raises(SystemExit): + convert_udm_to_ord.main(convert_udm_to_ord.parse_args(argv))