kg_microbe.transform_utils.rhea_mappings package

Submodules

kg_microbe.transform_utils.rhea_mappings.label_inputs module

Project exact external labels from consumed, prepared DBs in Rhea’s selected raw root.

kg_microbe.transform_utils.rhea_mappings.label_inputs.external_label_lookup(transform): Read immutable DB snapshots only; never build, download, remap IDs, or invent missing labels.

kg_microbe.transform_utils.rhea_mappings.label_inputs.validate_label_contract(recorded): Require an exact versioned reader contract, not Python’s Boolean/integer equality.

kg_microbe.transform_utils.rhea_mappings.label_inputs.verify_label_bindings(raw_dir, snapshots): Keep each consumed role rooted in the explicitly selected prepared authority.

kg_microbe.transform_utils.rhea_mappings.legacy_xrefs module

Audit six reviewed historical cross-references without restoring or remapping graph assertions.

kg_microbe.transform_utils.rhea_mappings.legacy_xrefs.load_legacy_xref_audit(transform, release_policy, native, *, policy_path=PosixPath('/home/runner/work/kg-microbe/kg-microbe/mappings/rhea_legacy_xref_dispositions.json'))

Consume the policy and return its zero or six scalar audit rows before publication.

The caller must already have loaded the native RDF through load_pinned_inputs. This checks that consumed RDF digest and selected release contract agree with this policy; it does not read historical RDF/tables, transform native records, rewrite references, or claim to revalidate comments or GO subclass evidence. Write all returned rows atomically with AUDIT_HEADER and record_producer_audit only after success; declare this module’s required input/audit/data constants.

kg_microbe.transform_utils.rhea_mappings.legacy_xrefs.read_legacy_policy(stream): Parse six reviewed exclusions or explicit absent history, never a general remapping language.

kg_microbe.transform_utils.rhea_mappings.native_graph module

Project validated native Rhea families without inferring flux or part/whole identity.

kg_microbe.transform_utils.rhea_mappings.native_graph.curated_xref(subject, target): Preserve the activity-cross-reference contract; this is not physical enablement.

kg_microbe.transform_utils.rhea_mappings.native_graph.edge(subject, predicate, target, relation=None, *, asserted=False, **witness): Distinguish native assertions from documented path-derived participation edges.

kg_microbe.transform_utils.rhea_mappings.native_graph.family_rows(family, native, label_lookup)

Yield (kind, row) for one validated family, retaining every source participant path.

Direct rh:chebi participants use ChEBI. Generic compounds and polymers keep their native identity: reactive parts and underlying ChEBI references are connected by their exact native relations, never substituted as reactants. UN and BI have undirected participation; only LR/RL get input/output roles.

kg_microbe.transform_utils.rhea_mappings.native_graph.native_curie(iri): Compact native nonnumeric resources without minting invalid numeric RHEA IDs.

kg_microbe.transform_utils.rhea_mappings.native_graph.native_xrefs(record): Read EC and GO from the same RDF snapshot, never independent latest tables.

kg_microbe.transform_utils.rhea_mappings.native_inputs module

Read a coherent, hash-pinned local Rhea snapshot without implicit cache downloads.

kg_microbe.transform_utils.rhea_mappings.native_inputs.load_pinned_inputs(transform, *, policy_path=PosixPath('/home/runner/work/kg-microbe/kg-microbe/mappings/rhea_release.json')): Read and cross-check all reaction families before any producer output is written.

kg_microbe.transform_utils.rhea_mappings.native_inputs.read_release_policy(stream): Validate the finite two-file pin contract, never selecting paths from input JSON.

kg_microbe.transform_utils.rhea_mappings.native_rdf module

Stream selected local Rhea RDF assertions without a graph loader or remote cache.

class kg_microbe.transform_utils.rhea_mappings.native_rdf.ChemicalParticipant(chebi_id, side_iri, participant_iri, compound_iri, route, reactive_part_iri=None)

Bases: object

Retain a ChEBI path witness without claiming whole-compound equivalence.

chebi_id: str

compound_iri: str

participant_iri: str

reactive_part_iri: str | None = None

route: str

side_iri: str

exception kg_microbe.transform_utils.rhea_mappings.native_rdf.NativeRheaError

Bases: ValueError

Signal missing, ambiguous or unsupported structure in selected native assertions.

class kg_microbe.transform_utils.rhea_mappings.native_rdf.NativeRheaIndex(reactions, sides, participants, compounds, reactive_parts, entity_metadata=<factory>)

Bases: object

Hold compact selected RDF adjacency, not a materialized XML or general RDF graph.

chemical_participants(side)

Resolve every direct, underlying-polymer or reactive-part path on one explicit side.

Return type:: tuple[ChemicalParticipant, ...]

compounds: dict[str, dict[str, tuple[str, ...]]]

entity_metadata: dict[str, dict[str, str]]

participants: dict[str, tuple[str, ...]]

reactions: dict[str, dict[str, tuple[str, ...]]]

reactive_parts: dict[str, tuple[str, ...]]

sides: dict[str, tuple[str, ...]]

kg_microbe.transform_utils.rhea_mappings.native_rdf.load_native_rdf(source, *, compressed=None)

Read only the supplied local path or caller-owned binary stream; never consult remote or home caches.

Return type:: NativeRheaIndex

kg_microbe.transform_utils.rhea_mappings.reaction_families module

Validate Rhea’s explicit four-member families without inferring reaction flux.

class kg_microbe.transform_utils.rhea_mappings.reaction_families.ReactionFamily(master, lr, rl, bi)

Bases: object

Four distinct canonical IDs whose roles come from one directions-table row.

bi: str

property links: tuple[tuple[str, str, str], ...]: Expose native RDF family links, not synthetic hierarchy or identity assertions.

lr: str

master: str

property members: tuple[tuple[str, str], ...]: Return source directions; UN is unspecified, never neutral or bidirectional.

node_annotations(identifier)

Return exact family annotations for one member, without changing its identity.

Return type:: dict[str, str]

rl: str

kg_microbe.transform_utils.rhea_mappings.reaction_families.load_reaction_families(text_stream)

Read the exact native TSV layout and validate all membership before returning.

Return type:: tuple[ReactionFamily, ...]

kg_microbe.transform_utils.rhea_mappings.reaction_families.validate_family_cohort(families, evidence)

Require directions membership to equal all active native RDF reaction IDs.

Only explicitly Obsolete records may lie outside the directions table; they need no labels or family structure and are returned for an upstream exclusion audit. Unknown, ambiguous, missing or mixed statuses fail. No native records are changed, and Preliminary is never called Approved.

Return type:: dict

kg_microbe.transform_utils.rhea_mappings.reaction_families.validate_family_evidence(family, evidence)

Require native RDF links, coherent active statuses, labels, and exact side roles.

Reaction keys are canonical RHEA CURIEs. Predicate keys and object values are full native IRIs, while labels are literal strings. The caller must combine all RDF descriptions of each subject before invoking this check. RDF duplicate triples are set-equivalent. Unrelated properties are retained outside this finite check; no labels, sides, statuses or links are repaired. Master side names follow the explicit native <master>_L/_R convention, not the numerical spacing or iteration order of reaction identifiers. Approved and Preliminary are both active native states and must agree across all four members; neither status is converted or promoted. This does not validate participants, stoichiometry, or physiological flux.

Return type:: None

kg_microbe.transform_utils.rhea_mappings.rhea_mappings module

Transform one pinned native Rhea snapshot, preserving explicit reaction directions.

class kg_microbe.transform_utils.rhea_mappings.rhea_mappings.RheaMappingsTransform(input_dir=None, output_dir=None)

Bases: Transform

Consume explicit local inputs; never infer directions from ID order or cache state.

DATA_INPUTS: tuple = ('mappings/rhea_release.json', 'mappings/rhea_legacy_xref_dispositions.json')

Repo-relative curation files this transform reads, beyond its own data/raw/ download.

Declared so freshness tooling can tell that an output is stale against its data rather than only its code. Without it a mapping correction lands, every consumer keeps reporting FRESH, and a re-merge silently ships the old groundings: #778 corrected 16 isolation-source ids and #786 rewrote the unified chemical SSSOM, and the merged KG built afterwards still asserted 75 organisms isolated from a “Cell Line”, because nothing re-ran the transforms that read those files (#812).

Paths are relative to the repo root. Keep them tracked in git — the freshness check uses commit time, not mtime, because git checkout rewrites mtimes without changing content (#797).

List every curation file read, not a representative one. A partial declaration fails silently and looks identical to a complete one: ontologies_stubs declared 1 of the 11 files it read and was reported fresh after changes to the other ten (#839). Where the set comes from a constant, derive this from it rather than restating it.

REQUIRED_AUDIT_FILES: tuple = ('rhea_reaction_dispositions.tsv', 'rhea_legacy_xref_dispositions.tsv'): Producer-owned disposition reports whose emitted bytes must survive finalization/merge.

REQUIRED_CONSUMED_INPUTS: tuple = ('rhea_release_manifest', 'rhea_directions', 'rhea_rdf', 'rhea_chebi_labels', 'rhea_go_labels', 'rhea_ec_labels', 'rhea_legacy_xref_dispositions'): Named generated inputs that must actually be read before fresh finalization. Unlike DATA_INPUTS, these are checked after upstream producers have run.

TRANSFORM_INPUTS: tuple = ('ontologies',): Finalization reads the ontologies output’s declarations, so the CHEBI nodes Rhea’s edges reference are not declared again here (#1511).

TSV_QUOTING = 3

property producer_native_inputs: Version the label projection; exact DB identities live in consumed-input roles.

run(data_file=None, show_status=True): Write canonical graph rows and audited statuses from coherent pinned native evidence.

verify_native_inputs(*, byte_verified=False, output_dir=None): Reject unconsumed SQLite sidecars even when each main-file SHA remains unchanged.

classmethod verify_recorded_native_inputs(report, report_path, *, admission=None): Enforce raw-root roles and quiescent SQLite state again at merge admission.

kg_microbe.transform_utils.rhea_mappings.stage_snapshot module

Explicitly stage a reviewed native Rhea pair; never repin or overwrite an existing snapshot.

kg_microbe.transform_utils.rhea_mappings.stage_snapshot.main(): Provide an explicit offline staging command for hash-pinned acquisition.

kg_microbe.transform_utils.rhea_mappings.stage_snapshot.stage_snapshot(source_dir, input_dir, *, policy_path=PosixPath('/home/runner/work/kg-microbe/kg-microbe/mappings/rhea_release.json'), source_archive=None)

Stage a complete local export under input_dir/rhea, or verify an identical existing snapshot.

The source is either a directory holding the pair (an explicitly chosen local export) or, with source_archive, the archived release tarball the policy names, as downloaded into the rhea_download quarantine. This function never performs network requests and never changes the reviewed pin. Differing or partial existing destinations are refused, not repaired or removed. The brief exclusive empty reservation is replaced by the fully validated directory, never by one payload at a time.

Module contents

RHEA mappings transform.

class kg_microbe.transform_utils.rhea_mappings.RheaMappingsTransform(input_dir=None, output_dir=None)

Bases: Transform

Consume explicit local inputs; never infer directions from ID order or cache state.

DATA_INPUTS: tuple = ('mappings/rhea_release.json', 'mappings/rhea_legacy_xref_dispositions.json')

Repo-relative curation files this transform reads, beyond its own data/raw/ download.

Declared so freshness tooling can tell that an output is stale against its data rather than only its code. Without it a mapping correction lands, every consumer keeps reporting FRESH, and a re-merge silently ships the old groundings: #778 corrected 16 isolation-source ids and #786 rewrote the unified chemical SSSOM, and the merged KG built afterwards still asserted 75 organisms isolated from a “Cell Line”, because nothing re-ran the transforms that read those files (#812).

Paths are relative to the repo root. Keep them tracked in git — the freshness check uses commit time, not mtime, because git checkout rewrites mtimes without changing content (#797).

List every curation file read, not a representative one. A partial declaration fails silently and looks identical to a complete one: ontologies_stubs declared 1 of the 11 files it read and was reported fresh after changes to the other ten (#839). Where the set comes from a constant, derive this from it rather than restating it.

REQUIRED_AUDIT_FILES: tuple = ('rhea_reaction_dispositions.tsv', 'rhea_legacy_xref_dispositions.tsv'): Producer-owned disposition reports whose emitted bytes must survive finalization/merge.

REQUIRED_CONSUMED_INPUTS: tuple = ('rhea_release_manifest', 'rhea_directions', 'rhea_rdf', 'rhea_chebi_labels', 'rhea_go_labels', 'rhea_ec_labels', 'rhea_legacy_xref_dispositions'): Named generated inputs that must actually be read before fresh finalization. Unlike DATA_INPUTS, these are checked after upstream producers have run.

TRANSFORM_INPUTS: tuple = ('ontologies',): Finalization reads the ontologies output’s declarations, so the CHEBI nodes Rhea’s edges reference are not declared again here (#1511).

TSV_QUOTING = 3

property producer_native_inputs: Version the label projection; exact DB identities live in consumed-input roles.

run(data_file=None, show_status=True): Write canonical graph rows and audited statuses from coherent pinned native evidence.

verify_native_inputs(*, byte_verified=False, output_dir=None): Reject unconsumed SQLite sidecars even when each main-file SHA remains unchanged.

classmethod verify_recorded_native_inputs(report, report_path, *, admission=None): Enforce raw-root roles and quiescent SQLite state again at merge admission.