kg_microbe.transform_utils.mediadive package

Submodules

kg_microbe.transform_utils.mediadive.bulk_inputs module

Bind MediaDive’s five required JSON reads to their original locators and bytes.

kg_microbe.transform_utils.mediadive.bulk_inputs.load_bulk_payload(transform, name, path): Parse and bind payload identity inside its actual immutable native-read boundary (#1299).

kg_microbe.transform_utils.mediadive.bulk_inputs.snapshot_bulk_input(transform, name, path): Guard the lexical path before and after consuming its immutable text snapshot.

kg_microbe.transform_utils.mediadive.bulk_inputs.verify_bulk_inputs(transform, *, byte_verified=False, output_dir=None): Verify this instance’s retained contract without replacing its original admission epoch.

kg_microbe.transform_utils.mediadive.bulk_inputs.verify_bulk_payloads(transform): Recheck every cached payload at run/finalization boundaries, never rebaseline its identity.

kg_microbe.transform_utils.mediadive.bulk_inputs.verify_bulk_record(transform, name, key): Verify one actual parsed record before lookup, without rehashing the complete source tree.

kg_microbe.transform_utils.mediadive.bulk_inputs.verify_recorded_bulk_inputs(report, report_path, *, admission=None): Validate recorded raw roles within the shared recorded-input admission boundary.

kg_microbe.transform_utils.mediadive.material_scope_audit module

Preserve reviewed material grounding candidates separately from identity.

class kg_microbe.transform_utils.mediadive.material_scope_audit.MaterialScopeAudit(transform, media_list)

Bases: object

Collect only the reviewed grounding candidates; never create or choose an identity.

observe(occurrence, raw): Attach available candidates to an actual emitted occurrence and its selected target.

write(): Write complete producer-time evidence, including a valid empty-cohort header.

kg_microbe.transform_utils.mediadive.material_scope_audit.p3556_material(raw): Recognize only an explicit supplier/product qualifier, never a name or source ID.

kg_microbe.transform_utils.mediadive.material_scope_audit.sugar_material(raw, name=None): Match only the reviewed source name and stock qualifier, with case/whitespace normalization.

kg_microbe.transform_utils.mediadive.material_scope_audit.verify_recorded_material_inputs(report, report_path): Require actual canonical evidence origins for the producer-bound candidate sidecar.

kg_microbe.transform_utils.mediadive.mediadive module

MediaDive KG.

Example script to transform downloaded data into a graph format that KGX can ingest directly, in either TSV or JSON format: https://github.com/NCATS-Tangerine/kgx/blob/master/data-preparation.md

Input: any file in data/raw/ (that was downloaded by placing a URL in incoming.txt/yaml and running run.py download.

Output: transformed data in data/raw/MediaDive:

Output these two files: - nodes.tsv - edges.tsv

class kg_microbe.transform_utils.mediadive.mediadive.MediaDiveTransform(input_dir=None, output_dir=None)

Bases: Transform

Template for how the transform class would be designed.

DATA_INPUTS: tuple = ('mappings/kgmicrobe_unified_entity_mappings.sssom.tsv.gz', 'mappings/mediadive_material_grounding_review.json', 'mappings/ingredient_context_dispositions.json', 'mappings/catalase_native_authority.json', 'mappings/catalase_historical_groundings.json', 'mappings/mediadive_recipe_identity_policy.json', 'mappings/mediadive_recipe_identity_evidence.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.

DEFAULT_INPUT_DIR = PosixPath('/home/runner/work/kg-microbe/kg-microbe/data/raw')

OPTIONAL_RAW_CONSUMED_INPUTS: tuple = (('micromediaparam_strict', 'compound_mappings_strict.tsv'), ('micromediaparam_hydrate', 'compound_mappings_strict_hydrate.tsv')): Named optional reads relative to the effective input_base_dir, including explicit absence. Kept separate so repository-relative optional inputs never silently relocate.

REQUIRED_AUDIT_FILES: tuple = ('mediadive_material_scope_quarantine.tsv', 'mediadive_nested_solution_identity.tsv', 'mediadive_missing_solution_references.tsv', 'mediadive_oatmeal_identity.tsv', 'mediadive_substrate_identity.tsv'): Producer-owned disposition reports whose emitted bytes must survive finalization/merge.

REQUIRED_CONSUMED_INPUTS: tuple = ('bacdive_taxon_lookup', 'mediadive_media_list', 'mediadive_media_detailed', 'mediadive_media_strains', 'mediadive_solutions', 'mediadive_compounds', 'material_scope_reviewed_claims', 'material_scope_policy', 'ingredient_material_context', 'ingredient_material_native_authority', 'catalase_historical_groundings', 'mediadive_recipe_identity_policy', 'mediadive_recipe_identity_evidence'): 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', 'bacdive'): Reads ontologies/chebi_nodes.tsv and chebi_edges.tsv (roles/categories) via constants.py (#1035), plus BacDive’s intermediate strain-taxid TSV (#1091).

consume_bulk_input(name, path): Consume a required JSON from the original lexical path and immutable bytes.

download_yaml_and_get_json(url, target_dir)

Fetch live MediaDive data; retain the legacy signature without YAML persistence.

Parameters:

url (str) – Path provided by MediaDive API.
target_dir (Path) – Ignored legacy cache destination; never read, created or written.

Return type:

Dict[str, str]

finalize(*, file_prefix='', fresh_run=False): Reconstruct a fresh instance’s native guards only from fully validated evidence.

get_compounds_of_solution(id)

Return the historical unique-display-name view, refusing lossy collisions.

Production emission uses get_solution_recipe_occurrences instead. This compatibility view is safe only when every display name is unique.

Parameters:: id (str) – ID of solution.
Returns:: Dictionary of display names to resolved identities and quantities.
Raises:: ValueError – Two recipe occurrences have the same display name.

get_json_object(fn, url_extension, target_dir)

Read admitted bulk data, or fetch live data for an explicit nonbulk lookup.

Bulk mode never falls back to YAML or HTTP for a missing record. Nonbulk helper calls are uncached diagnostics, not a production transform mode.

Parameters:

fn (Union[Path, str]) – Ignored legacy YAML file path.
url_extension (str) – API endpoint extension (e.g., “medium/123”).
target_dir (Path) – Ignored legacy YAML directory.

Return type:

Dict[str, str]

Returns:

Dictionary.

get_solution_recipe_occurrences(id)

Resolve every recipe occurrence in source-list order without name deduplication.

Assertion IDs use the owning solution and one-based raw list position, not the optional/nonunique source recipe_order. The exact source item is retained as canonical JSON evidence, including original name, source ID, source order, optional flag and all supplied quantity fields.

Parameters:: id (str) – ID of solution.
Returns:: Ordered occurrence dictionaries with resolved IDs and source evidence.

classmethod missing_prerequisites(input_dir=None)

Report an unusable ncit.db before the batch starts.

MediaDive’s nodes include reviewed NCIT category classes, so the NCIT category policy reads ncit.db when this source finalizes. It gets the ontologies source’s check: an existing copy is read as it is, so one left beside a newer ncit.db.gz would be read silently, and a damaged archive would fail only at finalization (#1536, #1557).

Parameters:: input_dir (Optional[Path]) – Effective raw input directory; None means the default.
Return type:: List[str]

property producer_native_inputs: Serialize original reads, never substitute current files at finalization.

run(data_file=None, show_status=True): Run the transformation, closing the API session afterwards.

standardize_compound_id(id, compound_name=None, *, source_record=None, source_assertion_id=None)

Get standardized IDs via unified chemical mappings, bulk data, or legacy mappings.

Lookup order: 1. Unified chemical mappings by compound name (ChemicalMappingLoader) 2. Legacy MicroMediaParam mappings by compound name (fallback during transition) 3. Bulk downloaded data (embedded ChEBI/KEGG/PubChem/CAS-RN) 4. Custom ingredient prefix (last resort)

Parameters:

id (str) – MediaDive compound ID.
compound_name (Optional[str]) – Compound name for mapping lookup.
source_record (Optional[dict]) – Actual occurrence evidence; even an empty dictionary overrides embedded qualifiers.

Returns:

Standardized ID.

verify_native_inputs(*, byte_verified=False, output_dir=None): Keep the constructor and selected list bound through finalization.

classmethod verify_recorded_native_inputs(report, report_path, *, admission=None): Admit the original raw JSON identities with the public source validator.

kg_microbe.transform_utils.mediadive.recipe_identity module

Preserve native recipe referents and six reviewed material observations (#1317–1319).

class kg_microbe.transform_utils.mediadive.recipe_identity.RecipeIdentityGuard(transform, media_list)

Bases: object

One immutable source-derived conservation index for a producer epoch.

compound_target(identifier, name, raw, assertion): Protect finite native compound records before unified, legacy or embedded lookup.

contract(): Return a detached record; only a sealed producer may claim completion.

observe(occurrence, raw): Record actual producer decisions without permitting a native coordinate collision.

restore(recorded): Reuse independently verified output without claiming a fresh writer run.

property sealed: Expose completion without permitting caller reset of evidence verification.

verify(*, output_dir=None, byte_verified=False): Check the original read epoch and, after sealing, the complete published evidence.

write(): Seal complete writer output and mandatory audits before generic finalization.

kg_microbe.transform_utils.mediadive.recipe_identity.digest(value): Hash the canonical evidence payload, not an outer file receipt.

kg_microbe.transform_utils.mediadive.recipe_identity.encoded(value): Preserve exact JSON values, positions and qualifiers in source evidence.

kg_microbe.transform_utils.mediadive.recipe_identity.native_id(value): Require an unambiguous positive source ID; booleans/floats are not identifiers.

kg_microbe.transform_utils.mediadive.recipe_identity.nested_identifier(raw): Explicit child identity precedes every lexical chemical mapping route.

kg_microbe.transform_utils.mediadive.recipe_identity.require(condition, message): Fail before publication when source identity or evidence cannot be established.

kg_microbe.transform_utils.mediadive.recipe_identity.restore_recipe_checkpoint(transform, report_path): Restore a fresh instance only through public validation and actual immutable rereads.

kg_microbe.transform_utils.mediadive.recipe_identity.verify_recorded_recipe_identity(report, report_path, *, admission): Independently reconstruct source decisions from retained, actually admitted native bytes.

Module contents

MediaDive transform.

class kg_microbe.transform_utils.mediadive.MediaDiveTransform(input_dir=None, output_dir=None)

Bases: Transform

Template for how the transform class would be designed.

DATA_INPUTS: tuple = ('mappings/kgmicrobe_unified_entity_mappings.sssom.tsv.gz', 'mappings/mediadive_material_grounding_review.json', 'mappings/ingredient_context_dispositions.json', 'mappings/catalase_native_authority.json', 'mappings/catalase_historical_groundings.json', 'mappings/mediadive_recipe_identity_policy.json', 'mappings/mediadive_recipe_identity_evidence.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.

DEFAULT_INPUT_DIR = PosixPath('/home/runner/work/kg-microbe/kg-microbe/data/raw')

OPTIONAL_RAW_CONSUMED_INPUTS: tuple = (('micromediaparam_strict', 'compound_mappings_strict.tsv'), ('micromediaparam_hydrate', 'compound_mappings_strict_hydrate.tsv')): Named optional reads relative to the effective input_base_dir, including explicit absence. Kept separate so repository-relative optional inputs never silently relocate.

REQUIRED_AUDIT_FILES: tuple = ('mediadive_material_scope_quarantine.tsv', 'mediadive_nested_solution_identity.tsv', 'mediadive_missing_solution_references.tsv', 'mediadive_oatmeal_identity.tsv', 'mediadive_substrate_identity.tsv'): Producer-owned disposition reports whose emitted bytes must survive finalization/merge.

REQUIRED_CONSUMED_INPUTS: tuple = ('bacdive_taxon_lookup', 'mediadive_media_list', 'mediadive_media_detailed', 'mediadive_media_strains', 'mediadive_solutions', 'mediadive_compounds', 'material_scope_reviewed_claims', 'material_scope_policy', 'ingredient_material_context', 'ingredient_material_native_authority', 'catalase_historical_groundings', 'mediadive_recipe_identity_policy', 'mediadive_recipe_identity_evidence'): 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', 'bacdive'): Reads ontologies/chebi_nodes.tsv and chebi_edges.tsv (roles/categories) via constants.py (#1035), plus BacDive’s intermediate strain-taxid TSV (#1091).

chebi_categories: Dict[str, str]

chebi_labels: Dict[str, str]

chebi_role_edges: Dict[str, list]

chebi_roles: Dict[str, list]

consume_bulk_input(name, path): Consume a required JSON from the original lexical path and immutable bytes.

download_yaml_and_get_json(url, target_dir)

Fetch live MediaDive data; retain the legacy signature without YAML persistence.

Parameters:

url (str) – Path provided by MediaDive API.
target_dir (Path) – Ignored legacy cache destination; never read, created or written.

Return type:

Dict[str, str]

finalize(*, file_prefix='', fresh_run=False): Reconstruct a fresh instance’s native guards only from fully validated evidence.

get_compounds_of_solution(id)

Return the historical unique-display-name view, refusing lossy collisions.

Production emission uses get_solution_recipe_occurrences instead. This compatibility view is safe only when every display name is unique.

Parameters:: id (str) – ID of solution.
Returns:: Dictionary of display names to resolved identities and quantities.
Raises:: ValueError – Two recipe occurrences have the same display name.

get_json_object(fn, url_extension, target_dir)

Read admitted bulk data, or fetch live data for an explicit nonbulk lookup.

Bulk mode never falls back to YAML or HTTP for a missing record. Nonbulk helper calls are uncached diagnostics, not a production transform mode.

Parameters:

fn (Union[Path, str]) – Ignored legacy YAML file path.
url_extension (str) – API endpoint extension (e.g., “medium/123”).
target_dir (Path) – Ignored legacy YAML directory.

Return type:

Dict[str, str]

Returns:

Dictionary.

get_solution_recipe_occurrences(id)

Resolve every recipe occurrence in source-list order without name deduplication.

Assertion IDs use the owning solution and one-based raw list position, not the optional/nonunique source recipe_order. The exact source item is retained as canonical JSON evidence, including original name, source ID, source order, optional flag and all supplied quantity fields.

Parameters:: id (str) – ID of solution.
Returns:: Ordered occurrence dictionaries with resolved IDs and source evidence.

classmethod missing_prerequisites(input_dir=None)

Report an unusable ncit.db before the batch starts.

MediaDive’s nodes include reviewed NCIT category classes, so the NCIT category policy reads ncit.db when this source finalizes. It gets the ontologies source’s check: an existing copy is read as it is, so one left beside a newer ncit.db.gz would be read silently, and a damaged archive would fail only at finalization (#1536, #1557).

Parameters:: input_dir (Optional[Path]) – Effective raw input directory; None means the default.
Return type:: List[str]

property producer_native_inputs: Serialize original reads, never substitute current files at finalization.

run(data_file=None, show_status=True): Run the transformation, closing the API session afterwards.

standardize_compound_id(id, compound_name=None, *, source_record=None, source_assertion_id=None)

Get standardized IDs via unified chemical mappings, bulk data, or legacy mappings.

Lookup order: 1. Unified chemical mappings by compound name (ChemicalMappingLoader) 2. Legacy MicroMediaParam mappings by compound name (fallback during transition) 3. Bulk downloaded data (embedded ChEBI/KEGG/PubChem/CAS-RN) 4. Custom ingredient prefix (last resort)

Parameters:

id (str) – MediaDive compound ID.
compound_name (Optional[str]) – Compound name for mapping lookup.
source_record (Optional[dict]) – Actual occurrence evidence; even an empty dictionary overrides embedded qualifiers.

Returns:

Standardized ID.

verify_native_inputs(*, byte_verified=False, output_dir=None): Keep the constructor and selected list bound through finalization.

classmethod verify_recorded_native_inputs(report, report_path, *, admission=None): Admit the original raw JSON identities with the public source validator.