diff --git a/README.md b/README.md index 494e096..14e6816 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,12 @@ For custom protocols, declare typed sample metadata with `protocol.add_sample(sa `lab.samples` also defines `Location(resource, well)`, `SamplePlacement`, and `OutputManifest`. Recorded operations, sample placements, target bindings, and final volume accounting use the same logical `Location` type. For example, an output placement's `location` can be used directly as a key in `dict(compiled.final_volumes)`. `lab.part.Part` identifies a biological part by its SBOL IRI; cloning types and stage builders live under `lab.experiments.cloning`. +## Biological designs and provenance + +`lab.provenance` provides immutable SBOL3 designs, material implementations, activities, qualified usages and associations, agents, and plans. Author a `Document`, explicitly add its objects, and call `freeze()` to validate references and obtain a reproducible snapshot. Import and export local Turtle with `Document.read()` and `snapshot.write()`, or exchange a detached pySBOL3 document with `from_sbol3()` and `to_sbol3()`. + +See the [provenance guide](docs/provenance.md) and run `uv run --no-sync python -m examples.provenance` for an authoring and round-trip example. + ## Describe a deck This OT-2 deck places two 96-well plates in slots 1 and 2, a 300 µL tip rack in slot 3, and a P300 pipette on the left mount. It uses the same equipment and placement types as the [deck layouts example](https://github.com/the-lab-compiler/lab-py/blob/master/examples/deck_layouts.py). diff --git a/docs/provenance.md b/docs/provenance.md new file mode 100644 index 0000000..c0cd42e --- /dev/null +++ b/docs/provenance.md @@ -0,0 +1,123 @@ +# Biological designs and provenance + +`lab.provenance` represents SBOL3 designs and the activities, agents, plans, and material realizations associated with them. Its objects are immutable Python dataclasses. A `Document` collects explicitly added objects, and `freeze()` produces a validated `DocumentSnapshot` with stable identities for owned children. The module imports and exports SBOL3 Turtle and interoperates with pySBOL3 without changing its global namespace or builder registry. + +## Author a document + +```python +from lab.provenance import Component, Document, Sequence +from lab.provenance.vocabulary import DNA, IUPAC_DNA + +document = Document(namespace="https://example.org/my_project") +sequence = Sequence( + identity=document.iri("sequence"), + elements="ACGTACGT", + encoding=IUPAC_DNA, +) +design = Component( + identity=document.iri("design"), + types=(DNA,), + sequences=(sequence.ref,), +) +document.add(sequence, design) +snapshot = document.freeze() +snapshot.write("build/design.ttl") +``` + +`Ref[T]` carries an absolute IRI and a Python target type. `design.ref` is a `Ref[Component]`; `snapshot.resolve(design.ref)` returns the corresponding component. `snapshot.get(identity, Component)` also checks the requested type at runtime. References never fetch remote resources. Adding an object does not automatically add its references. + +Collections use tuples. SBOL multi-valued properties are unordered RDF sets, so freezing sorts them deterministically. Represent biological order with locations and constraints. Tuple position does not specify an assembly recipe. + +`document.iri("design/feature")` constructs an IRI from slash-separated SBOL display IDs. It does not use a process-wide namespace. Top-level objects require explicit identities. Owned children such as `Usage`, `Association`, and `SubComponent` may omit theirs; freezing assigns names such as `planned_build/Usage1` in a new snapshot. Explicit child identities are preserved, and generated identities avoid existing ones. Give a child an explicit identity when another object needs to reference it before freezing. + +`Document.add()` accepts top-level objects and rejects conflicting definitions of an existing IRI. Adding an identical definition is idempotent. Edits use `dataclasses.replace()` and a new identity where they describe a new design or record. `Document.from_snapshot(snapshot)` creates an independent authoring document containing an existing snapshot's objects and annotations. + +## Model designs, materials, and activities + +| Objects | Purpose | +| --- | --- | +| `Sequence`, `Component` | Sequence data and structural or functional designs | +| `SubComponent`, `SequenceFeature`, `LocalSubComponent`, `ExternallyDefined`, `ComponentReference` | Features owned by a component | +| `Range`, `Cut`, `EntireSequence` | Locations on a referenced sequence | +| `Constraint`, `Interaction`, `Participation`, `Interface` | Structural relationships and functional interactions | +| `CombinatorialDerivation`, `VariableFeature`, `Collection` | Design families and grouped objects | +| `Implementation` | A planned, recorded, or simulated material realization | +| `Activity`, `Usage`, `Association` | Work, qualified input roles, and qualified agent roles | +| `Agent`, `Plan` | Who or what is involved, and the method identity | +| `Attachment`, `ExperimentalData`, `Experiment`, `Model`, `Measure` | Linked evidence, datasets, computational models, and quantities | + +Objects share `name`, `description`, `derived_from`, `generated_by`, and owned `measures`. Top-level objects also have `namespace` and attachment references. `Usage` and `Association` belong to an `Activity`; agents, plans, activities, and implementations are top-level objects. + +```python +from lab.provenance import ( + Activity, + Agent, + AgentKind, + Association, + EvidenceState, + Implementation, + Plan, + Usage, +) + +planner = Agent( + identity=document.iri("planner"), + kind=AgentKind.SOFTWARE, + software_version="0.1.0", +) +method = Plan(identity=document.iri("method")) +activity = Activity( + identity=document.iri("planned_build"), + usage=(Usage(entity=design.ref),), + association=(Association(agent=planner.ref, plan=method.ref),), + evidence_state=EvidenceState.PLANNED, +) +output = Implementation( + identity=document.iri("planned_output"), + derived_from=(design.ref,), + generated_by=(activity.ref,), + evidence_state=EvidenceState.PLANNED, +) +document.add(planner, method, activity, output) +``` + +`Implementation.derived_from` can identify the intended design. `built` describes the realized structure when that assertion is available. Planned implementations must leave `built` unset. An implementation's material inputs belong in the generating activity's usages; the model does not infer genetic lineage from every physical reagent contribution. + +Evidence states are `UNKNOWN`, `PLANNED`, `RECORDED`, and `SIMULATED`. A recorded inventory assertion does not imply sequence verification or a successful experiment. Imported SBOL without an evidence state remains unknown. Planned activities cannot have execution timestamps. Recorded and simulated activities can include timezone-aware `datetime` values, and end times must not precede start times. + +`Activity.types` contains ontology classifications encoded as `sbol:type`. `Activity.informed_by` references predecessor activities through `prov:wasInformedBy`; multiple activities can share a predecessor. `Plan.protocol` optionally links to a separately specified protocol IRI, including a LabOP protocol. The plan itself does not contain or execute a method body. + +## Import, validate, and export + +```python +document = Document.read("build/design.ttl") +document.validate().raise_for_errors() +snapshot = document.freeze() + +native = snapshot.to_sbol3() # A new, mutable pySBOL3 Document +native_report = native.validate() # pySBOL3's additional SBOL/SHACL checks +restored = Document.from_sbol3(native).freeze() +assert restored.digest == snapshot.digest +``` + +Input is local SBOL3 Turtle. SBOL2 input and unsupported SBOL classes or properties fail explicitly. Foreign RDF annotations, including language-tagged values and nested blank nodes, survive import and export. Unknown annotations remain RDF rather than becoming inferred Python fields. A mixed-namespace or empty import needs an explicit `namespace=` for authoring new objects. + +Lab checks field types, required values, unique identities and ownership, reference closure and target types, sequence bounds, feature reference scope, component containment and activity dependency cycles, and evidence assertions. `validate()` returns diagnostics with identity, field path, code, and message. `freeze()` raises `ProvenanceError` containing that report when checks fail. These checks are not a complete implementation of every SBOL specification rule; use the detached pySBOL3 document's validator for additional checks. + +Use `freeze(allow_external=True)` for intentionally incomplete graphs. References to objects that are present still undergo type validation; unresolved references remain unresolved. The module never creates placeholder objects or downloads referenced attachments. + +Turtle output is deterministic, with full IRIs and canonical blank-node identifiers. `snapshot.digest` is SHA-256 over that serialization, including retained annotations. This identifies the provenance document, independently of the compiler artifact digest. `write()` accepts an identical existing file and refuses to replace different contents. + +The adapter pins pySBOL3 `1.2.0.post0`. It handles `Activity.informed_by` at the RDF boundary because the upstream property uses ownership. It also corrects that release's mappings of `Cut.at` and `SubComponent.role_integration` on detached returned instances, while preserving the standard `sbol:at` and `sbol:roleIntegration` predicates. Importing an existing pySBOL3 document normalizes those known mappings without mutating the original. No global classes, namespaces, or builders are patched. + +The small Lab extension vocabulary is packaged at `lab/provenance/resources/lab.ttl`. It defines evidence state, agent kind, software version, and the link from a plan to a protocol. The namespace is an identifier; using it does not require a network lookup. + +## Run the example + +```sh +uv run --no-sync python -m examples.provenance --out build/provenance.ttl +``` + +[The example](../examples/provenance.py) builds a design, an inventory assertion, a prospective activity, and a planned output; checks the SBOL3 export; writes it; and verifies its round trip. It specifies no executable cloning method. + +The mappings follow the [SBOL 3.1 specification](https://sbolstandard.org/docs/SBOL3.1.0.pdf) and [pySBOL3's provenance model](https://raw.githubusercontent.com/SynBioDex/pySBOL3/main/sbol3/provenance.py). diff --git a/examples/provenance.py b/examples/provenance.py new file mode 100644 index 0000000..bf8ae2f --- /dev/null +++ b/examples/provenance.py @@ -0,0 +1,94 @@ +"""Author and round-trip an SBOL3 design and prospective build provenance. + +Run ``python -m examples.provenance --out build/provenance.ttl``. The short +sequence is illustrative; this example specifies no executable cloning method. +""" + +import argparse +from pathlib import Path + +from lab import __version__ +from lab.provenance import ( + Activity, + Agent, + AgentKind, + Association, + Component, + Document, + EvidenceState, + Implementation, + Plan, + Sequence, + SubComponent, + Usage, +) +from lab.provenance.vocabulary import DNA, IUPAC_DNA, LAB + + +def build_document() -> Document: + document = Document(namespace="https://example.org/lab/provenance_example") + sequence = Sequence( + identity=document.iri("input_sequence"), + elements="ACGTACGT", + encoding=IUPAC_DNA, + ) + part = Component( + identity=document.iri("input_design"), + name="Illustrative input design", + types=(DNA,), + sequences=(sequence.ref,), + ) + product = Component( + identity=document.iri("product_design"), + name="Intended product design", + types=(DNA,), + features=(SubComponent(instance_of=part.ref),), + ) + stock = Implementation( + identity=document.iri("input_stock"), + derived_from=(part.ref,), + evidence_state=EvidenceState.RECORDED, + description="An author-supplied inventory assertion; sequence verification unspecified.", + ) + planner = Agent( + identity=document.iri("lab_compiler"), + name="Lab Compiler", + kind=AgentKind.SOFTWARE, + software_version=__version__, + ) + method = Plan( + identity=document.iri("method"), + description="Prospective method identity; detailed protocol specification is separate.", + ) + activity = Activity( + identity=document.iri("planned_build"), + evidence_state=EvidenceState.PLANNED, + usage=(Usage(entity=stock.ref, roles=(LAB + "inputMaterial",)),), + association=(Association(agent=planner.ref, plan=method.ref, roles=(LAB + "planner",)),), + ) + output = Implementation( + identity=document.iri("planned_output"), + derived_from=(product.ref,), + generated_by=(activity.ref,), + evidence_state=EvidenceState.PLANNED, + ) + document.add(sequence, part, product, stock, planner, method, activity, output) + return document + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", type=Path, default=Path("build/provenance.ttl")) + args = parser.parse_args() + frozen = build_document().freeze() + report = frozen.to_sbol3().validate() + if report.errors: + raise ValueError(str(report)) + output = frozen.write(args.out) + restored = Document.read(output).freeze() + assert restored.digest == frozen.digest + print(f"Wrote {output}; {len(frozen.objects)} top-level objects; SHA-256 {frozen.digest}") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 2b0a7ee..9403ad6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,11 @@ classifiers = [ "Topic :: Scientific/Engineering", "Typing :: Typed", ] -dependencies = ["pint>=0.24,<0.27"] +dependencies = [ + "pint>=0.24,<0.27", + "sbol3==1.2.0.post0", + "rdflib>=6.1.1,<7", +] [project.urls] Homepage = "https://github.com/the-lab-compiler/lab-py" @@ -87,5 +91,5 @@ check_untyped_defs = true disallow_untyped_defs = true [[tool.mypy.overrides]] -module = ["opentrons.*", "pylabrobot.*"] +module = ["opentrons.*", "pylabrobot.*", "sbol3.*"] ignore_missing_imports = true diff --git a/scripts/check_install.py b/scripts/check_install.py index d3bfb05..72d9d2b 100644 --- a/scripts/check_install.py +++ b/scripts/check_install.py @@ -9,6 +9,7 @@ from tempfile import TemporaryDirectory import lab +from lab.provenance import Activity, Document from lab.targets import Manual @@ -17,12 +18,14 @@ def main() -> None: import_module("lab.experiments.cloning") import_module("lab.part") import_module("lab.samples") + import_module("lab.provenance") package = distribution("lab-compiler") assert package.version == lab.__version__ assert package.metadata["Name"] == "lab-compiler" assert set(package.metadata.get_all("Provides-Extra", [])) == {"opentrons", "star"} assert files("lab").joinpath("py.typed").is_file() + assert files("lab.provenance").joinpath("resources/lab.ttl").is_file() assert any(str(path).endswith("licenses/LICENSE") for path in package.files or ()) assert lab.__file__ is not None assert not Path(lab.__file__).resolve().is_relative_to(Path(__file__).resolve().parents[1]) @@ -40,6 +43,13 @@ def main() -> None: assert plan["compiler_version"] == package.version assert (output / "protocol.html").stat().st_size > 0 + provenance = Document(namespace="https://example.org/install_check") + provenance.add(Activity(identity=provenance.iri("activity"))) + snapshot = provenance.freeze() + snapshot.write(output / "provenance.ttl") + assert Document.read(output / "provenance.ttl").freeze().digest == snapshot.digest + assert not snapshot.to_sbol3().validate().errors + assert not any(name.split(".")[0] in {"opentrons", "pylabrobot"} for name in sys.modules) print(f"lab-compiler {package.version}: installed package check passed") diff --git a/src/lab/provenance/__init__.py b/src/lab/provenance/__init__.py new file mode 100644 index 0000000..0755933 --- /dev/null +++ b/src/lab/provenance/__init__.py @@ -0,0 +1,83 @@ +"""SBOL3-native biological designs and planned or recorded provenance.""" + +from lab.provenance.document import Document, DocumentSnapshot +from lab.provenance.types import ( + Activity, + Agent, + Association, + Attachment, + Collection, + CombinatorialDerivation, + Component, + ComponentReference, + Constraint, + Cut, + EntireSequence, + Experiment, + ExperimentalData, + ExternallyDefined, + Feature, + Identified, + Implementation, + Interaction, + Interface, + LocalSubComponent, + Measure, + Model, + Participation, + Plan, + Range, + Ref, + Sequence, + SequenceFeature, + SequenceLocation, + SubComponent, + TopLevel, + Usage, + VariableFeature, +) +from lab.provenance.validation import Diagnostic, ProvenanceError, ValidationReport +from lab.provenance.vocabulary import AgentKind, EvidenceState + +__all__ = [ + "Activity", + "Agent", + "AgentKind", + "Association", + "Attachment", + "Collection", + "CombinatorialDerivation", + "Component", + "ComponentReference", + "Constraint", + "Cut", + "Diagnostic", + "Document", + "DocumentSnapshot", + "EntireSequence", + "EvidenceState", + "Experiment", + "ExperimentalData", + "ExternallyDefined", + "Feature", + "Identified", + "Implementation", + "Interaction", + "Interface", + "LocalSubComponent", + "Measure", + "Model", + "Participation", + "Plan", + "ProvenanceError", + "Range", + "Ref", + "Sequence", + "SequenceFeature", + "SequenceLocation", + "SubComponent", + "TopLevel", + "Usage", + "ValidationReport", + "VariableFeature", +] diff --git a/src/lab/provenance/_schema.py b/src/lab/provenance/_schema.py new file mode 100644 index 0000000..9aa8182 --- /dev/null +++ b/src/lab/provenance/_schema.py @@ -0,0 +1,229 @@ +"""Explicit SBOL3 property mapping shared by validation and serialization.""" + +from dataclasses import dataclass +from typing import Literal + +from lab.provenance.types import ( + Activity, + Agent, + Association, + Attachment, + Collection, + CombinatorialDerivation, + Component, + ComponentReference, + Constraint, + Cut, + EntireSequence, + Experiment, + ExperimentalData, + ExternallyDefined, + Feature, + Identified, + Implementation, + Interaction, + Interface, + LocalSubComponent, + Measure, + Model, + Participation, + Plan, + Range, + Sequence, + SequenceFeature, + SequenceLocation, + SubComponent, + TopLevel, + Usage, + VariableFeature, +) +from lab.provenance.vocabulary import LAB, OM, PROV, SBOL, AgentKind, EvidenceState + +Kind = Literal["text", "iri", "reference", "owned", "integer", "float", "datetime", "enum"] + + +@dataclass(frozen=True) +class Property: + predicate: str + kind: Kind + multiple: bool = False + required: bool = False + target: type | None = None + + +# These mappings are intentionally explicit: Python attribute names are not RDF +# predicate names, and Activity.types is sbol:type, not rdf:type. +PROPERTIES: dict[type[Identified], dict[str, Property]] = { + Identified: { + "name": Property(SBOL + "name", "text"), + "description": Property(SBOL + "description", "text"), + "derived_from": Property(PROV + "wasDerivedFrom", "reference", True, target=Identified), + "generated_by": Property(PROV + "wasGeneratedBy", "reference", True, target=Activity), + "measures": Property(SBOL + "hasMeasure", "owned", True, target=Measure), + }, + TopLevel: { + "namespace": Property(SBOL + "hasNamespace", "iri"), + "attachments": Property(SBOL + "hasAttachment", "reference", True, target=Attachment), + }, + Sequence: { + "elements": Property(SBOL + "elements", "text", required=True), + "encoding": Property(SBOL + "encoding", "iri", required=True), + }, + Component: { + "types": Property(SBOL + "type", "iri", True, True), + "roles": Property(SBOL + "role", "iri", True), + "sequences": Property(SBOL + "hasSequence", "reference", True, target=Sequence), + "features": Property(SBOL + "hasFeature", "owned", True, target=Feature), + "constraints": Property(SBOL + "hasConstraint", "owned", True, target=Constraint), + "interactions": Property(SBOL + "hasInteraction", "owned", True, target=Interaction), + "interface": Property(SBOL + "hasInterface", "owned", target=Interface), + "models": Property(SBOL + "hasModel", "reference", True, target=Model), + }, + Feature: { + "roles": Property(SBOL + "role", "iri", True), + "orientation": Property(SBOL + "orientation", "iri"), + }, + SubComponent: { + "instance_of": Property(SBOL + "instanceOf", "reference", required=True, target=Component), + "role_integration": Property(SBOL + "roleIntegration", "iri"), + "locations": Property(SBOL + "hasLocation", "owned", True, target=SequenceLocation), + "source_locations": Property( + SBOL + "sourceLocation", "owned", True, target=SequenceLocation + ), + }, + SequenceFeature: { + "locations": Property(SBOL + "hasLocation", "owned", True, True, SequenceLocation), + }, + LocalSubComponent: { + "types": Property(SBOL + "type", "iri", True, True), + "locations": Property(SBOL + "hasLocation", "owned", True, target=SequenceLocation), + }, + ExternallyDefined: { + "types": Property(SBOL + "type", "iri", True, True), + "definition": Property(SBOL + "definition", "iri", required=True), + }, + ComponentReference: { + "in_child_of": Property( + SBOL + "inChildOf", "reference", required=True, target=SubComponent + ), + "refers_to": Property(SBOL + "refersTo", "reference", required=True, target=Feature), + }, + SequenceLocation: { + "sequence": Property(SBOL + "hasSequence", "reference", required=True, target=Sequence), + "orientation": Property(SBOL + "orientation", "iri"), + "order": Property(SBOL + "order", "integer"), + }, + Range: { + "start": Property(SBOL + "start", "integer", required=True), + "end": Property(SBOL + "end", "integer", required=True), + }, + Cut: {"at": Property(SBOL + "at", "integer", required=True)}, + EntireSequence: {}, + Constraint: { + "restriction": Property(SBOL + "restriction", "iri", required=True), + "subject": Property(SBOL + "subject", "reference", required=True, target=Feature), + "object": Property(SBOL + "object", "reference", required=True, target=Feature), + }, + Interaction: { + "types": Property(SBOL + "type", "iri", True, True), + "participations": Property(SBOL + "hasParticipation", "owned", True, target=Participation), + }, + Participation: { + "roles": Property(SBOL + "role", "iri", True, True), + "participant": Property(SBOL + "participant", "reference", required=True, target=Feature), + }, + Interface: { + "inputs": Property(SBOL + "input", "reference", True, target=Feature), + "outputs": Property(SBOL + "output", "reference", True, target=Feature), + "nondirectionals": Property(SBOL + "nondirectional", "reference", True, target=Feature), + }, + Implementation: { + "built": Property(SBOL + "built", "reference", target=Component), + "evidence_state": Property(LAB + "evidenceState", "enum", target=EvidenceState), + }, + Attachment: { + "source": Property(SBOL + "source", "iri", required=True), + "format": Property(SBOL + "format", "iri"), + "size": Property(SBOL + "size", "integer"), + "hash": Property(SBOL + "hash", "text"), + "hash_algorithm": Property(SBOL + "hashAlgorithm", "text"), + }, + Model: { + "source": Property(SBOL + "source", "iri", required=True), + "language": Property(SBOL + "language", "iri", required=True), + "framework": Property(SBOL + "framework", "iri", required=True), + }, + Collection: {"members": Property(SBOL + "member", "reference", True, target=TopLevel)}, + Experiment: {"members": Property(SBOL + "member", "reference", True, target=ExperimentalData)}, + ExperimentalData: {}, + CombinatorialDerivation: { + "template": Property(SBOL + "template", "reference", required=True, target=Component), + "strategy": Property(SBOL + "strategy", "iri"), + "variable_features": Property( + SBOL + "hasVariableFeature", "owned", True, target=VariableFeature + ), + }, + VariableFeature: { + "cardinality": Property(SBOL + "cardinality", "iri", required=True), + "variable": Property(SBOL + "variable", "reference", required=True, target=Feature), + "variants": Property(SBOL + "variant", "reference", True, target=Component), + "variant_collections": Property( + SBOL + "variantCollection", "reference", True, target=Collection + ), + "variant_derivations": Property( + SBOL + "variantDerivation", "reference", True, target=CombinatorialDerivation + ), + "variant_measures": Property(SBOL + "variantMeasure", "owned", True, target=Measure), + }, + Measure: { + "value": Property(OM + "hasNumericalValue", "float", required=True), + "unit": Property(OM + "hasUnit", "iri", required=True), + "types": Property(SBOL + "type", "iri", True), + }, + Usage: { + "entity": Property(PROV + "entity", "reference", required=True, target=Identified), + "roles": Property(PROV + "hadRole", "iri", True), + }, + Association: { + "agent": Property(PROV + "agent", "reference", required=True, target=Agent), + "plan": Property(PROV + "hadPlan", "reference", target=Plan), + "roles": Property(PROV + "hadRole", "iri", True), + }, + Agent: { + "kind": Property(LAB + "agentKind", "enum", target=AgentKind), + "software_version": Property(LAB + "softwareVersion", "text"), + }, + Plan: {"protocol": Property(LAB + "protocol", "iri")}, + Activity: { + "types": Property(SBOL + "type", "iri", True), + "usage": Property(PROV + "qualifiedUsage", "owned", True, target=Usage), + "association": Property(PROV + "qualifiedAssociation", "owned", True, target=Association), + "informed_by": Property(PROV + "wasInformedBy", "reference", True, target=Activity), + "start_time": Property(PROV + "startedAtTime", "datetime"), + "end_time": Property(PROV + "endedAtTime", "datetime"), + "evidence_state": Property(LAB + "evidenceState", "enum", target=EvidenceState), + }, +} + +ABSTRACT = {Identified, TopLevel, Feature, SequenceLocation} +RDF_TYPES = { + cls: ( + PROV + if cls in {Activity, Agent, Plan, Usage, Association} + else OM + if cls is Measure + else SBOL + ) + + cls.__name__ + for cls in PROPERTIES + if cls not in ABSTRACT +} + + +def properties(cls: type[Identified]) -> dict[str, Property]: + if cls not in RDF_TYPES: + raise TypeError(f"Unsupported provenance class: {cls.__name__}") + result: dict[str, Property] = {} + for base in reversed(cls.__mro__): + result.update(PROPERTIES.get(base, {})) + return result diff --git a/src/lab/provenance/document.py b/src/lab/provenance/document.py new file mode 100644 index 0000000..43611f3 --- /dev/null +++ b/src/lab/provenance/document.py @@ -0,0 +1,262 @@ +"""Explicit document ownership, immutable snapshots, and local interchange.""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Any, TypeVar, cast + +import sbol3 as native + +from lab.provenance import sbol3 as adapter +from lab.provenance._schema import properties +from lab.provenance.types import Identified, Ref, TopLevel, require_iri +from lab.provenance.validation import ( + ProvenanceError, + ValidationReport, + validate_objects, + validate_structure, + values, + walk, +) + +T = TypeVar("T", bound=Identified) + + +def _find(objects: tuple[TopLevel, ...], identity: str) -> Identified: + for top in objects: + for obj in walk(top): + if obj.identity == identity: + return obj + raise KeyError(identity) + + +def _normalized(objects: tuple[TopLevel, ...], namespace: str) -> tuple[TopLevel, ...]: + validate_structure(objects).raise_for_errors() + reserved = {obj.identity for top in objects for obj in walk(top) if obj.identity is not None} + + def normalize(obj: Identified, identity: str) -> Identified: + changes: dict[str, Any] = {"identity": identity} + if isinstance(obj, TopLevel) and obj.namespace is None: + if identity.startswith(namespace): + changes["namespace"] = namespace + else: + # Imported/mixed namespaces remain explicit, independent of the + # document's namespace for new authoring. + changes["namespace"] = identity.rsplit("/", 1)[0] if "/" in identity else namespace + counts: dict[str, int] = {} + for name, prop in properties(type(obj)).items(): + items = values(obj, name, prop) + if not items: + continue + if prop.kind == "owned": + children: list[Identified] = [] + for child in items: + assert isinstance(child, Identified) + child_id = child.identity + if child_id is None: + label = type(child).__name__ + number = counts.get(label, 0) + 1 + child_id = f"{identity}/{label}{number}" + while child_id in reserved: + number += 1 + child_id = f"{identity}/{label}{number}" + counts[label] = number + reserved.add(child_id) + children.append(normalize(child, child_id)) + changes[name] = ( + tuple(sorted(children, key=lambda child: child.identity or "")) + if prop.multiple + else children[0] + ) + elif prop.multiple: + # SBOL multi-valued properties are RDF sets. Biological order is + # encoded by locations and constraints, never tuple position. + changes[name] = tuple( + sorted( + items, + key=lambda item: item.identity if isinstance(item, Ref) else str(item), + ) + ) + return replace(obj, **changes) + + return tuple( + cast(TopLevel, normalize(obj, obj.identity)) + for obj in sorted(objects, key=lambda obj: obj.identity) + ) + + +def _write(path: str | Path, text: str) -> Path: + path = Path(path) + if path.exists() and path.read_text(encoding="utf-8") != text: + raise FileExistsError(f"{path} already contains a different artifact") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +@dataclass(frozen=True, kw_only=True) +class DocumentSnapshot: + namespace: str + objects: tuple[TopLevel, ...] + allow_external: bool = False + _extra_rdf: str = "" + + def __post_init__(self) -> None: + require_iri(self.namespace) + if not isinstance(self.objects, tuple): + raise TypeError("Snapshot objects must be a tuple") + self.validate().raise_for_errors() + + def get(self, identity: str, expected: type[T]) -> T: + obj = _find(self.objects, identity) + if not isinstance(obj, expected): + raise TypeError(f"{identity} is {type(obj).__name__}, not {expected.__name__}") + return obj + + def resolve(self, ref: Ref[T]) -> T: + return cast(T, _find(self.objects, ref.identity)) + + def validate(self) -> ValidationReport: + return validate_objects(self.objects, allow_external=self.allow_external) + + def to_sbol3(self) -> native.Document: + """Return a new mutable pySBOL3 document, detached from this snapshot.""" + return adapter.to_sbol3(self.objects, self._extra_rdf) + + def to_turtle(self) -> str: + return adapter.serialize(self.objects, self._extra_rdf) + + @property + def digest(self) -> str: + return hashlib.sha256(self.to_turtle().encode("utf-8")).hexdigest() + + def write(self, path: str | Path) -> Path: + return _write(path, self.to_turtle()) + + +class Document: + """An explicit collection of SBOL top-level objects. + + Adding objects does not add their referenced objects, infer missing designs, + fetch URLs, or change a global namespace. ``freeze`` assigns identities to + owned children and checks reference closure before returning a snapshot. + """ + + def __init__(self, *, namespace: str) -> None: + self.namespace = require_iri(namespace) + self._objects: dict[str, TopLevel] = {} + self._extra_rdf = "" + + def iri(self, local_id: str) -> str: + if not isinstance(local_id, str) or not re.fullmatch( + r"[A-Za-z_][A-Za-z0-9_]*(/[A-Za-z_][A-Za-z0-9_]*)*", local_id + ): + raise ValueError( + "Use slash-separated SBOL display IDs containing letters, digits, or underscores" + ) + return self.namespace.rstrip("/") + "/" + local_id + + @property + def objects(self) -> tuple[TopLevel, ...]: + return tuple(self._objects.values()) + + def add(self, *objects: TopLevel) -> None: + """Add top-level objects atomically; an identical addition is idempotent.""" + validate_structure(objects).raise_for_errors() + updated = dict(self._objects) + for obj in objects: + existing = updated.get(obj.identity) + if ( + existing is not None + and existing != obj + and _normalized((existing,), self.namespace) != _normalized((obj,), self.namespace) + ): + raise ValueError(f"Conflicting definition for {obj.identity}") + updated[obj.identity] = existing or obj + self._objects = updated + + def replace(self, *objects: TopLevel) -> None: + """Explicitly revise existing definitions in this authoring document. + + Frozen snapshots remain unchanged. Identity and concrete type must already + exist; freeze validates references and evidence after the revisions. + """ + validate_structure(objects).raise_for_errors() + updated = dict(self._objects) + for obj in objects: + if obj.identity not in updated: + raise KeyError(obj.identity) + if type(updated[obj.identity]) is not type(obj): + raise TypeError("Replacing a definition cannot change its concrete type") + updated[obj.identity] = obj + self._objects = updated + + def get(self, identity: str, expected: type[T]) -> T: + obj = _find(self.objects, identity) + if not isinstance(obj, expected): + raise TypeError(f"{identity} is {type(obj).__name__}, not {expected.__name__}") + return obj + + def resolve(self, ref: Ref[T]) -> T: + return cast(T, _find(self.objects, ref.identity)) + + def validate(self, *, allow_external: bool = False) -> ValidationReport: + try: + objects = _normalized(self.objects, self.namespace) + except ProvenanceError as error: + return error.report + return validate_objects(objects, allow_external=allow_external) + + def freeze(self, *, allow_external: bool = False) -> DocumentSnapshot: + return DocumentSnapshot( + namespace=self.namespace, + objects=_normalized(self.objects, self.namespace), + allow_external=allow_external, + _extra_rdf=self._extra_rdf, + ) + + def to_sbol3(self, *, allow_external: bool = False) -> native.Document: + return self.freeze(allow_external=allow_external).to_sbol3() + + def write(self, path: str | Path, *, allow_external: bool = False) -> Path: + return self.freeze(allow_external=allow_external).write(path) + + @classmethod + def from_snapshot(cls, snapshot: DocumentSnapshot) -> Document: + result = cls(namespace=snapshot.namespace) + result.add(*snapshot.objects) + result._extra_rdf = snapshot._extra_rdf + return result + + @classmethod + def _from_objects( + cls, objects: tuple[TopLevel, ...], extra_rdf: str, namespace: str | None + ) -> Document: + if namespace is None: + namespaces = {obj.namespace for obj in objects if obj.namespace is not None} + if len(namespaces) != 1: + raise ValueError("Supply namespace= for an empty document or multiple namespaces") + namespace = namespaces.pop() + assert namespace is not None + result = cls(namespace=namespace) + result.add(*objects) + result._extra_rdf = extra_rdf + return result + + @classmethod + def from_sbol3(cls, document: native.Document, *, namespace: str | None = None) -> Document: + objects, extra = adapter.from_sbol3(document) + return cls._from_objects(objects, extra, namespace) + + @classmethod + def from_turtle(cls, text: str, *, namespace: str | None = None) -> Document: + objects, extra = adapter.parse(text) + return cls._from_objects(objects, extra, namespace) + + @classmethod + def read(cls, path: str | Path, *, namespace: str | None = None) -> Document: + """Read local Turtle; referenced resources are never fetched.""" + return cls.from_turtle(Path(path).read_text(encoding="utf-8"), namespace=namespace) diff --git a/src/lab/provenance/resources/lab.ttl b/src/lab/provenance/resources/lab.ttl new file mode 100644 index 0000000..a9efafe --- /dev/null +++ b/src/lab/provenance/resources/lab.ttl @@ -0,0 +1,18 @@ +@prefix lab: . +@prefix rdf: . +@prefix rdfs: . + +lab:evidenceState a rdf:Property ; + rdfs:comment "Nature of the assertion; recorded does not imply sequence verification or experimental success." . +lab:unknown rdfs:label "Evidence state unspecified" . +lab:planned rdfs:label "Prospective work or output" . +lab:recorded rdfs:label "Recorded activity or material assertion" . +lab:simulated rdfs:label "Simulated activity or output" . +lab:agentKind a rdf:Property ; + rdfs:comment "PROV Person, Organization, or SoftwareAgent classification." . +lab:softwareVersion a rdf:Property ; + rdfs:comment "Version of the software represented by this agent." . +lab:protocol a rdf:Property ; + rdfs:comment "IRI of the protocol specified by a PROV Plan, for example a LabOP Protocol." . +lab:inputMaterial rdfs:label "Material used as an input to an activity" . +lab:planner rdfs:label "Agent responsible for planning an activity" . diff --git a/src/lab/provenance/sbol3.py b/src/lab/provenance/sbol3.py new file mode 100644 index 0000000..645e46a --- /dev/null +++ b/src/lab/provenance/sbol3.py @@ -0,0 +1,242 @@ +"""SBOL3 RDF interchange and a detached pySBOL3 adapter. + +RDF is the interoperability boundary. Constructing the graph explicitly preserves +owned identities and avoids pySBOL3's Activity.informed_by reparenting behavior. +No builder registrations or process-wide namespaces are changed. +""" + +from datetime import datetime +from enum import Enum +from typing import Any + +import sbol3 as native +from rdflib import RDF, XSD, BNode, Graph, Literal, URIRef +from rdflib.compare import to_canonical_graph +from rdflib.term import Identifier, Node +from sbol3.identified import extract_display_id, is_valid_display_id + +from lab.provenance._schema import RDF_TYPES, Property, properties +from lab.provenance.types import Identified, Ref, TopLevel, require_iri +from lab.provenance.validation import values, walk +from lab.provenance.vocabulary import SBOL, EvidenceState + + +def _n3(term: Node) -> str: + assert isinstance(term, (URIRef, BNode, Literal)) + return term.n3() + + +def canonical_rdf(graph: Graph) -> str: + """Deterministic Turtle using full IRIs and canonical blank-node identifiers.""" + canonical = to_canonical_graph(graph) + return "".join(sorted(f"{_n3(s)} {_n3(p)} {_n3(o)} .\n" for s, p, o in canonical)) + + +def graph_for(objects: tuple[TopLevel, ...], extra_rdf: str = "") -> Graph: + graph = Graph() + if extra_rdf: + graph.parse(data=extra_rdf, format="turtle") + for top in objects: + for obj in walk(top): + assert obj.identity is not None + subject = URIRef(obj.identity) + graph.add((subject, RDF.type, URIRef(RDF_TYPES[type(obj)]))) + if not RDF_TYPES[type(obj)].startswith(SBOL): + base = "TopLevel" if isinstance(obj, TopLevel) else "Identified" + graph.add((subject, RDF.type, URIRef(SBOL + base))) + # Preserve an imported displayId; otherwise derive it as pySBOL3 does. + if not list(graph.objects(subject, URIRef(SBOL + "displayId"))): + display_id = extract_display_id(obj.identity) + if display_id is not None: + graph.add((subject, URIRef(SBOL + "displayId"), Literal(display_id))) + for name, prop in properties(type(obj)).items(): + for value in values(obj, name, prop): + # Absence means unknown; do not fabricate assertions on import. + if value is EvidenceState.UNKNOWN: + continue + if isinstance(value, (Ref, Identified)): + assert value.identity is not None + term: Identifier = URIRef(value.identity) + elif isinstance(value, Enum): + term = URIRef(value.value) + elif prop.kind == "iri": + term = URIRef(str(value)) + elif prop.kind == "float": + term = Literal(float(value), datatype=XSD.float) # type: ignore[arg-type] + else: + term = Literal(value) + graph.add((subject, URIRef(prop.predicate), term)) + return graph + + +def serialize(objects: tuple[TopLevel, ...], extra_rdf: str = "") -> str: + return canonical_rdf(graph_for(objects, extra_rdf)) + + +def parse(data: str, *, format: str = "turtle") -> tuple[tuple[TopLevel, ...], str]: + """Parse local RDF data; formats that can fetch remote contexts are excluded.""" + if format not in {"turtle", "nt"}: + raise ValueError("Provenance input must be Turtle or N-Triples") + # A relative IRI without an explicit @base must not silently become a path + # inside the caller's current working directory. + graph = Graph().parse(data=data, format=format, publicID="urn:lab:provenance:input") + by_type: dict[Node, type[Identified]] = {URIRef(iri): cls for cls, iri in RDF_TYPES.items()} + classes: dict[Node, type[Identified]] = {} + for subject, _, rdf_type in graph.triples((None, RDF.type, None)): + if str(rdf_type).startswith("http://sbols.org/v2#"): + raise ValueError("SBOL2 input is unsupported; supply an SBOL3 document") + if rdf_type not in by_type: + if str(rdf_type) in {SBOL + "TopLevel", SBOL + "Identified"}: + continue + if str(rdf_type).startswith(SBOL): + raise ValueError(f"Unsupported SBOL class {rdf_type}") + continue + if not isinstance(subject, URIRef): + raise ValueError("SBOL objects need absolute IRI identities, not blank nodes") + if subject in classes and classes[subject] is not by_type[rdf_type]: + raise ValueError(f"Conflicting SBOL classes for {subject}") + classes[subject] = by_type[rdf_type] + + consumed = Graph() + cache: dict[Node, Identified] = {} + visiting: set[Node] = set() + owners: dict[Node, Node] = {} + + def decode(term: Node, prop: Property) -> object: + if prop.kind in {"iri", "reference", "owned", "enum"}: + if not isinstance(term, URIRef): + raise ValueError(f"{prop.predicate} requires an IRI, got {term!r}") + require_iri(str(term)) + if prop.kind == "reference": + return Ref(str(term)) + if prop.kind == "owned": + return build(term) + if prop.kind == "enum": + assert prop.target is not None + return prop.target(str(term)) + return str(term) + if not isinstance(term, Literal): + raise ValueError(f"{prop.predicate} requires a literal, got {term!r}") + python = term.toPython() + if prop.kind == "text": + if term.language or (term.datatype is not None and term.datatype != XSD.string): + raise ValueError(f"{prop.predicate} requires an untagged string literal") + return str(term) + if prop.kind == "integer" and type(python) is int: + return python + if prop.kind == "float" and type(python) in {float, int}: + return float(python) + if prop.kind == "datetime" and isinstance(python, datetime): + return python + raise ValueError(f"Invalid {prop.kind} literal for {prop.predicate}: {term!r}") + + def build(subject: Node) -> Identified: + if subject in visiting: + raise ValueError(f"Cyclic SBOL ownership at {subject}") + if subject in cache: + return cache[subject] + cls = classes.get(subject) + if cls is None: + raise ValueError(f"Missing or unsupported owned object {subject}") + visiting.add(subject) + args: dict[str, Any] = {"identity": str(subject)} + consumed.add((subject, RDF.type, URIRef(RDF_TYPES[cls]))) + for name, prop in properties(cls).items(): + predicate = URIRef(prop.predicate) + terms = sorted(graph.objects(subject, predicate), key=_n3) + if not prop.multiple and len(terms) > 1: + raise ValueError(f"{subject}.{name} must have at most one value") + if prop.required and not terms: + raise ValueError(f"{subject}.{name} is required") + if prop.kind == "owned": + for term in terms: + if term in owners: + raise ValueError(f"Owned object {term} has more than one owner/property") + owners[term] = subject + if terms: + decoded = tuple(decode(term, prop) for term in terms) + args[name] = decoded if prop.multiple else decoded[0] + for term in terms: + consumed.add((subject, predicate, term)) + result = cls(**args) + cache[subject] = result + visiting.remove(subject) + return result + + objects = tuple( + build(subject) + for subject in sorted(classes, key=str) + if issubclass(classes[subject], TopLevel) + ) + for subject, cls in classes.items(): + if not issubclass(cls, TopLevel) and subject not in owners: + raise ValueError(f"Orphaned owned object {subject}") + for subject, _, _ in graph.triples((None, URIRef(SBOL + "hasNamespace"), None)): + if subject not in classes: + raise ValueError(f"Unsupported top-level object {subject}") + extra = graph - consumed + for subject in classes: + display_id = extract_display_id(str(subject)) + declared = tuple(graph.objects(subject, URIRef(SBOL + "displayId"))) + if len(declared) > 1: + raise ValueError(f"{subject}.displayId must have at most one value") + if declared: + value = declared[0] + if ( + not isinstance(value, Literal) + or value.language + or value.datatype not in {None, XSD.string} + or not is_valid_display_id(str(value)) + or (display_id is not None and str(value) != display_id) + ): + raise ValueError(f"Invalid displayId for {subject}: {value!r}") + if display_id is not None: + extra.remove((subject, URIRef(SBOL + "displayId"), None)) + if not RDF_TYPES[classes[subject]].startswith(SBOL): + base = "TopLevel" if issubclass(classes[subject], TopLevel) else "Identified" + extra.remove((subject, RDF.type, URIRef(SBOL + base))) + # Foreign annotation graphs, including nested blank nodes, are retained. An + # unrecognized SBOL property is an unsupported schema, not an annotation. + for subject, predicate, _ in extra: + if str(predicate).startswith(SBOL) and predicate != URIRef(SBOL + "displayId"): + raise ValueError(f"Unsupported SBOL property {predicate} on {subject}") + assert all(isinstance(obj, TopLevel) for obj in objects) + return tuple(obj for obj in objects if isinstance(obj, TopLevel)), canonical_rdf(extra) + + +def to_sbol3(objects: tuple[TopLevel, ...], extra_rdf: str = "") -> native.Document: + document = native.Document() + document.read_string(serialize(objects, extra_rdf), native.TURTLE) + # Two further mapping defects in the pinned pySBOL3 release are repaired on + # these detached instances only. Standard predicates remain intact in RDF. + for top in objects: + for obj in walk(top): + converted = document.find(obj.identity) + if isinstance(converted, native.Cut): + object.__setattr__( + converted, "at", native.IntProperty(converted, SBOL + "at", 1, 1) + ) + if isinstance(converted, native.SubComponent): + object.__setattr__( + converted, + "role_integration", + native.URIProperty(converted, SBOL + "roleIntegration", 0, 1), + ) + return document + + +def from_sbol3(document: native.Document) -> tuple[tuple[TopLevel, ...], str]: + if not isinstance(document, native.Document): + raise TypeError("Pass a pySBOL3 Document") + graph = document.graph() + # Normalize only the known legacy predicates emitted by this pySBOL3 version. + for subject in graph.subjects(RDF.type, URIRef(SBOL + "Cut")): + for value in tuple(graph.objects(subject, URIRef(SBOL + "start"))): + graph.remove((subject, URIRef(SBOL + "start"), value)) + graph.add((subject, URIRef(SBOL + "at"), value)) + for subject in graph.subjects(RDF.type, URIRef(SBOL + "SubComponent")): + for value in tuple(graph.objects(subject, URIRef(SBOL + "role"))): + if str(value) in {SBOL + "mergeRoles", SBOL + "overrideRoles"}: + graph.remove((subject, URIRef(SBOL + "role"), value)) + graph.add((subject, URIRef(SBOL + "roleIntegration"), value)) + return parse(canonical_rdf(graph)) diff --git a/src/lab/provenance/types.py b/src/lab/provenance/types.py new file mode 100644 index 0000000..832739a --- /dev/null +++ b/src/lab/provenance/types.py @@ -0,0 +1,290 @@ +"""Immutable SBOL3 design and provenance types, with no global document state.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, fields +from datetime import datetime +from typing import Generic, Self, TypeVar +from urllib.parse import urlsplit + +from lab.provenance.vocabulary import AgentKind, EvidenceState + +T_co = TypeVar("T_co", bound="Identified", covariant=True) + + +def require_iri(value: str) -> str: + """Require an absolute IRI; never resolve, fetch, or rewrite it.""" + if ( + not isinstance(value, str) + or not value + or re.search(r'[\s<>"{}|\\^`\x00-\x1f\x7f]', value) + or not re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", value) + ): + raise ValueError(f"Expected an absolute IRI, got {value!r}") + parsed = urlsplit(value) + if parsed.scheme in {"http", "https"} and not parsed.netloc: + raise ValueError(f"Expected an absolute IRI, got {value!r}") + return value + + +@dataclass(frozen=True, slots=True) +class Ref(Generic[T_co]): + identity: str + + def __post_init__(self) -> None: + require_iri(self.identity) + + +@dataclass(frozen=True, kw_only=True) +class Identified: + """Common SBOL metadata. Only owned objects may omit an identity.""" + + identity: str | None = None + name: str | None = None + description: str | None = None + derived_from: tuple[Ref[Identified], ...] = () + generated_by: tuple[Ref[Activity], ...] = () + measures: tuple[Measure, ...] = () + + def __post_init__(self) -> None: + if self.identity is not None: + require_iri(self.identity) + # Reject shallowly frozen objects containing mutable collections. + for field in fields(self): + value = getattr(self, field.name) + if isinstance(value, (list, dict, set)): + raise TypeError( + f"{type(self).__name__}.{field.name} must be immutable; use a tuple" + ) + + @property + def ref(self) -> Ref[Self]: + if self.identity is None: + raise ValueError("Assign an identity, or retrieve the owned object after freezing") + return Ref(self.identity) + + +@dataclass(frozen=True, kw_only=True) +class TopLevel(Identified): + identity: str + namespace: str | None = None + attachments: tuple[Ref[Attachment], ...] = () + + def __post_init__(self) -> None: + super().__post_init__() + require_iri(self.identity) + if self.namespace is not None: + require_iri(self.namespace) + + +@dataclass(frozen=True, kw_only=True) +class Measure(Identified): + value: float + unit: str + types: tuple[str, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Sequence(TopLevel): + elements: str + encoding: str + + +@dataclass(frozen=True, kw_only=True) +class SequenceLocation(Identified): + sequence: Ref[Sequence] + orientation: str | None = None + order: int | None = None + + +@dataclass(frozen=True, kw_only=True) +class Range(SequenceLocation): + """One-based, inclusive sequence coordinates.""" + + start: int + end: int + + +@dataclass(frozen=True, kw_only=True) +class Cut(SequenceLocation): + """Position between bases; zero denotes the beginning of a sequence.""" + + at: int + + +@dataclass(frozen=True, kw_only=True) +class EntireSequence(SequenceLocation): + pass + + +@dataclass(frozen=True, kw_only=True) +class Feature(Identified): + roles: tuple[str, ...] = () + orientation: str | None = None + + +@dataclass(frozen=True, kw_only=True) +class SubComponent(Feature): + instance_of: Ref[Component] + role_integration: str | None = None + locations: tuple[SequenceLocation, ...] = () + source_locations: tuple[SequenceLocation, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class SequenceFeature(Feature): + locations: tuple[SequenceLocation, ...] + + +@dataclass(frozen=True, kw_only=True) +class LocalSubComponent(Feature): + types: tuple[str, ...] + locations: tuple[SequenceLocation, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class ExternallyDefined(Feature): + types: tuple[str, ...] + definition: str + + +@dataclass(frozen=True, kw_only=True) +class ComponentReference(Feature): + in_child_of: Ref[SubComponent] + refers_to: Ref[Feature] + + +@dataclass(frozen=True, kw_only=True) +class Constraint(Identified): + restriction: str + subject: Ref[Feature] + object: Ref[Feature] + + +@dataclass(frozen=True, kw_only=True) +class Participation(Identified): + roles: tuple[str, ...] + participant: Ref[Feature] + + +@dataclass(frozen=True, kw_only=True) +class Interaction(Identified): + types: tuple[str, ...] + participations: tuple[Participation, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Interface(Identified): + inputs: tuple[Ref[Feature], ...] = () + outputs: tuple[Ref[Feature], ...] = () + nondirectionals: tuple[Ref[Feature], ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Component(TopLevel): + types: tuple[str, ...] + roles: tuple[str, ...] = () + sequences: tuple[Ref[Sequence], ...] = () + features: tuple[Feature, ...] = () + constraints: tuple[Constraint, ...] = () + interactions: tuple[Interaction, ...] = () + interface: Interface | None = None + models: tuple[Ref[Model], ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Implementation(TopLevel): + """A planned, recorded, or simulated realization of a design. + + ``derived_from`` names the intended design. ``built`` describes the realized + structure, when known. A planned output does not assert that it was built. + """ + + built: Ref[Component] | None = None + evidence_state: EvidenceState = EvidenceState.UNKNOWN + + +@dataclass(frozen=True, kw_only=True) +class Attachment(TopLevel): + source: str + format: str | None = None + size: int | None = None + hash: str | None = None + hash_algorithm: str | None = None + + +@dataclass(frozen=True, kw_only=True) +class Model(TopLevel): + source: str + language: str + framework: str + + +@dataclass(frozen=True, kw_only=True) +class Collection(TopLevel): + members: tuple[Ref[TopLevel], ...] = () + + +@dataclass(frozen=True, kw_only=True) +class ExperimentalData(TopLevel): + pass + + +@dataclass(frozen=True, kw_only=True) +class Experiment(TopLevel): + members: tuple[Ref[ExperimentalData], ...] = () + + +@dataclass(frozen=True, kw_only=True) +class VariableFeature(Identified): + cardinality: str + variable: Ref[Feature] + variants: tuple[Ref[Component], ...] = () + variant_collections: tuple[Ref[Collection], ...] = () + variant_derivations: tuple[Ref[CombinatorialDerivation], ...] = () + variant_measures: tuple[Measure, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class CombinatorialDerivation(TopLevel): + template: Ref[Component] + strategy: str | None = None + variable_features: tuple[VariableFeature, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Usage(Identified): + entity: Ref[Identified] + roles: tuple[str, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Association(Identified): + agent: Ref[Agent] + plan: Ref[Plan] | None = None + roles: tuple[str, ...] = () + + +@dataclass(frozen=True, kw_only=True) +class Agent(TopLevel): + kind: AgentKind | None = None + software_version: str | None = None + + +@dataclass(frozen=True, kw_only=True) +class Plan(TopLevel): + """Method identity; ``protocol`` links to its separately specified protocol.""" + + protocol: str | None = None + + +@dataclass(frozen=True, kw_only=True) +class Activity(TopLevel): + types: tuple[str, ...] = () + usage: tuple[Usage, ...] = () + association: tuple[Association, ...] = () + informed_by: tuple[Ref[Activity], ...] = () + start_time: datetime | None = None + end_time: datetime | None = None + evidence_state: EvidenceState = EvidenceState.UNKNOWN diff --git a/src/lab/provenance/validation.py b/src/lab/provenance/validation.py new file mode 100644 index 0000000..3bee4be --- /dev/null +++ b/src/lab/provenance/validation.py @@ -0,0 +1,373 @@ +"""Structural, reference, and evidence checks for Lab provenance documents.""" + +import math +from dataclasses import dataclass +from datetime import datetime +from urllib.parse import urlsplit + +from sbol3.identified import extract_display_id + +from lab.provenance._schema import Property, properties +from lab.provenance.types import ( + Activity, + Attachment, + CombinatorialDerivation, + Component, + ComponentReference, + Cut, + Identified, + Implementation, + Range, + Ref, + Sequence, + SequenceLocation, + SubComponent, + TopLevel, + VariableFeature, + require_iri, +) +from lab.provenance.vocabulary import INLINE, REVERSE_COMPLEMENT, SBOL, EvidenceState + + +@dataclass(frozen=True) +class Diagnostic: + identity: str | None + path: str + code: str + message: str + + +@dataclass(frozen=True) +class ValidationReport: + errors: tuple[Diagnostic, ...] = () + + @property + def is_valid(self) -> bool: + return not self.errors + + def raise_for_errors(self) -> None: + if self.errors: + raise ProvenanceError(self) + + +class ProvenanceError(ValueError): + def __init__(self, report: ValidationReport) -> None: + self.report = report + super().__init__( + "\n".join( + f"{item.identity or ''}.{item.path}: {item.message} [{item.code}]" + for item in report.errors + ) + ) + + +def values(obj: Identified, name: str, prop: Property) -> tuple[object, ...]: + value = getattr(obj, name) + if prop.multiple: + return value if isinstance(value, tuple) else (value,) + return () if value is None else (value,) + + +def owned(obj: Identified) -> tuple[Identified, ...]: + return tuple( + value + for name, prop in properties(type(obj)).items() + if prop.kind == "owned" + for value in values(obj, name, prop) + if isinstance(value, Identified) + ) + + +def walk(obj: Identified) -> tuple[Identified, ...]: + return (obj, *(descendant for child in owned(obj) for descendant in walk(child))) + + +def _valid_value(value: object, prop: Property) -> bool: + if prop.kind == "reference": + return isinstance(value, Ref) + if prop.kind in {"owned", "enum"}: + return prop.target is not None and isinstance(value, prop.target) + if prop.kind == "integer": + return type(value) is int + if prop.kind == "float": + return type(value) in {float, int} and math.isfinite(value) # type: ignore[arg-type] + if prop.kind == "datetime": + return isinstance(value, datetime) + if not isinstance(value, str): + return False + if prop.kind == "iri": + try: + require_iri(value) + except ValueError: + return False + return True + + +def validate_structure(objects: tuple[TopLevel, ...]) -> ValidationReport: + """Check field shapes before ownership normalization or RDF serialization.""" + errors: list[Diagnostic] = [] + + def visit(obj: Identified) -> None: + try: + schema = properties(type(obj)) + except TypeError as error: + errors.append(Diagnostic(obj.identity, "type", "unsupported-type", str(error))) + return + for name, prop in schema.items(): + raw = getattr(obj, name) + items = values(obj, name, prop) + if prop.multiple and not isinstance(raw, tuple): + errors.append(Diagnostic(obj.identity, name, "field-type", "Expected a tuple")) + if prop.required and not items: + errors.append(Diagnostic(obj.identity, name, "required", "A value is required")) + for value in items: + if not _valid_value(value, prop): + errors.append( + Diagnostic( + obj.identity, name, "field-type", f"Invalid {prop.kind} value {value!r}" + ) + ) + elif prop.kind == "owned" and isinstance(value, Identified): + visit(value) + if ( + prop.multiple + and prop.kind != "owned" + and all(_valid_value(value, prop) for value in items) + and len(set(items)) != len(items) + ): + errors.append( + Diagnostic( + obj.identity, + name, + "duplicate-value", + "SBOL properties contain unique values", + ) + ) + + for obj in objects: + if not isinstance(obj, TopLevel): + errors.append(Diagnostic(None, "objects", "field-type", "Expected a TopLevel object")) + else: + visit(obj) + return ValidationReport(tuple(errors)) + + +def validate_objects( + objects: tuple[TopLevel, ...], *, allow_external: bool = False +) -> ValidationReport: + structural = validate_structure(objects) + if not structural.is_valid: + return structural + errors: list[Diagnostic] = [] + index: dict[str, Identified] = {} + + def error(obj: Identified, path: str, code: str, message: str) -> None: + errors.append(Diagnostic(obj.identity, path, code, message)) + + for top in objects: + for obj in walk(top): + if obj.identity is None: + error( + obj, "identity", "missing-identity", "Freeze the document to assign identities" + ) + elif obj.identity in index: + error(obj, "identity", "duplicate-identity", "An identity has more than one owner") + else: + index[obj.identity] = obj + for obj in index.values(): + try: + extract_display_id(obj.identity) + except ValueError as invalid_identity: + error(obj, "identity", "display-id", str(invalid_identity)) + if ( + isinstance(obj, TopLevel) + and obj.namespace is not None + and urlsplit(obj.identity).netloc + and not obj.identity.startswith(obj.namespace) + ): + error(obj, "namespace", "namespace", "Namespace must be a prefix of the identity") + for name, prop in properties(type(obj)).items(): + if prop.kind != "reference": + continue + for value in values(obj, name, prop): + assert isinstance(value, Ref) + target = index.get(value.identity) + if target is None and not allow_external: + error(obj, name, "unresolved-reference", f"Missing {value.identity}") + elif ( + target is not None + and prop.target is not None + and not isinstance(target, prop.target) + ): + error( + obj, + name, + "reference-type", + f"{value.identity} must reference {prop.target.__name__}", + ) + orientation = getattr(obj, "orientation", None) + if orientation is not None and orientation not in {INLINE, REVERSE_COMPLEMENT}: + error(obj, "orientation", "orientation", "Expected inline or reverseComplement") + if ( + isinstance(obj, SubComponent) + and obj.role_integration is not None + and obj.role_integration not in {SBOL + "mergeRoles", SBOL + "overrideRoles"} + ): + error( + obj, "role_integration", "role-integration", "Expected mergeRoles or overrideRoles" + ) + if isinstance(obj, SequenceLocation): + if obj.order is not None and obj.order < 1: + error(obj, "order", "location-order", "Order must be positive") + sequence = index.get(obj.sequence.identity) + length = len(sequence.elements) if isinstance(sequence, Sequence) else None + if isinstance(obj, Range) and ( + obj.start < 1 or obj.end < obj.start or (length is not None and obj.end > length) + ): + error( + obj, + "start/end", + "sequence-range", + "Range must be within its sequence (1-based, inclusive)", + ) + if isinstance(obj, Cut) and (obj.at < 0 or (length is not None and obj.at > length)): + error(obj, "at", "sequence-cut", "Cut must be between zero and sequence length") + if isinstance(obj, Attachment): + if obj.size is not None and obj.size < 0: + error(obj, "size", "attachment-size", "Size must not be negative") + if (obj.hash is None) != (obj.hash_algorithm is None): + error(obj, "hash", "attachment-hash", "Supply both hash and hash_algorithm") + if ( + isinstance(obj, Implementation) + and obj.evidence_state is EvidenceState.PLANNED + and obj.built is not None + ): + error( + obj, + "built", + "planned-realization", + "Use derived_from for the intended design of a planned output", + ) + if isinstance(obj, Activity): + times = tuple(t for t in (obj.start_time, obj.end_time) if t is not None) + aware = all(t.tzinfo is not None and t.utcoffset() is not None for t in times) + if not aware: + error(obj, "start_time/end_time", "timezone", "Timestamps must include a timezone") + if ( + aware + and obj.start_time is not None + and obj.end_time is not None + and obj.end_time < obj.start_time + ): + error(obj, "end_time", "time-order", "End precedes start") + if times and obj.evidence_state is EvidenceState.PLANNED: + error( + obj, + "start_time/end_time", + "planned-execution", + "Planned activities cannot assert execution times", + ) + if isinstance(obj, VariableFeature) and obj.cardinality not in { + SBOL + word for word in ("one", "zeroOrOne", "oneOrMore", "zeroOrMore") + }: + error(obj, "cardinality", "cardinality", "Unknown SBOL cardinality") + if isinstance(obj, CombinatorialDerivation): + if obj.strategy is not None and obj.strategy not in { + SBOL + "sample", + SBOL + "enumerate", + }: + error(obj, "strategy", "strategy", "Unknown SBOL derivation strategy") + template = index.get(obj.template.identity) + if isinstance(template, Component): + members = {feature.identity for feature in template.features} + for variable in obj.variable_features: + if variable.variable.identity not in members: + error( + variable, + "variable", + "feature-scope", + "Variable must belong to the template", + ) + if isinstance(obj, Component): + members = {feature.identity for feature in obj.features} + scoped: list[tuple[Identified, str, Ref[Identified]]] = [ + (constraint, name, ref) + for constraint in obj.constraints + for name, ref in (("subject", constraint.subject), ("object", constraint.object)) + ] + scoped.extend( + (p, "participant", p.participant) + for interaction in obj.interactions + for p in interaction.participations + ) + if obj.interface is not None: + scoped.extend( + (obj.interface, name, ref) + for name in ("inputs", "outputs", "nondirectionals") + for ref in getattr(obj.interface, name) + ) + for scoped_child, name, ref in scoped: + if ref.identity not in members: + error( + scoped_child, name, "feature-scope", "Feature must belong to this component" + ) + for feature in obj.features: + if isinstance(feature, ComponentReference): + child = index.get(feature.in_child_of.identity) + child_component = ( + index.get(child.instance_of.identity) + if isinstance(child, SubComponent) + else None + ) + if feature.in_child_of.identity not in members: + error( + feature, + "in_child_of", + "feature-scope", + "SubComponent must belong to this component", + ) + if isinstance( + child_component, Component + ) and feature.refers_to.identity not in { + item.identity for item in child_component.features + }: + error( + feature, + "refers_to", + "feature-scope", + "Feature must belong to the referenced component", + ) + + # A temporal dependency graph and a component containment graph must be acyclic. + def check_cycles(cls: type[Activity] | type[Component], field: str) -> None: + visited: set[str] = set() + visiting: set[str] = set() + + def visit(identity: str) -> None: + obj = index.get(identity) + if not isinstance(obj, cls) or identity in visited: + return + if identity in visiting: + error(obj, field, "dependency-cycle", "Cyclic dependency") + return + visiting.add(identity) + refs = ( + obj.informed_by + if isinstance(obj, Activity) + else tuple( + feature.instance_of + for feature in obj.features + if isinstance(feature, SubComponent) + ) + ) + for ref in refs: + visit(ref.identity) + visiting.remove(identity) + visited.add(identity) + + for identity in index: + visit(identity) + + check_cycles(Activity, "informed_by") + check_cycles(Component, "features") + return ValidationReport(tuple(errors)) diff --git a/src/lab/provenance/vocabulary.py b/src/lab/provenance/vocabulary.py new file mode 100644 index 0000000..0aa2dc8 --- /dev/null +++ b/src/lab/provenance/vocabulary.py @@ -0,0 +1,38 @@ +"""SBOL, PROV, and Lab terms used by the public provenance model. + +Ontology terms remain ordinary absolute IRIs, so callers can use other ontologies +without registering Python classes or changing a process-wide namespace. +""" + +from enum import StrEnum + +SBOL = "http://sbols.org/v3#" +PROV = "http://www.w3.org/ns/prov#" +OM = "http://www.ontology-of-units-of-measure.org/resource/om-2/" +LAB = "https://the-lab-compiler.github.io/lab-py/ns#" + +DNA = "https://identifiers.org/SBO:0000251" +RNA = "https://identifiers.org/SBO:0000250" +PROTEIN = "https://identifiers.org/SBO:0000252" +SMALL_MOLECULE = "https://identifiers.org/SBO:0000247" +FUNCTIONAL_ENTITY = "https://identifiers.org/SBO:0000241" +IUPAC_DNA = "https://identifiers.org/edam:format_1207" +IUPAC_PROTEIN = "https://identifiers.org/edam:format_1208" +INLINE = SBOL + "inline" +REVERSE_COMPLEMENT = SBOL + "reverseComplement" +PRECEDES = SBOL + "precedes" + + +class EvidenceState(StrEnum): + """Nature of an assertion, independent of experimental verification.""" + + UNKNOWN = LAB + "unknown" + PLANNED = LAB + "planned" + RECORDED = LAB + "recorded" + SIMULATED = LAB + "simulated" + + +class AgentKind(StrEnum): + PERSON = PROV + "Person" + ORGANIZATION = PROV + "Organization" + SOFTWARE = PROV + "SoftwareAgent" diff --git a/tests/fixtures/provenance/design.ttl b/tests/fixtures/provenance/design.ttl new file mode 100644 index 0000000..78b8f90 --- /dev/null +++ b/tests/fixtures/provenance/design.ttl @@ -0,0 +1,27 @@ +@prefix sbol: . +@prefix ex: . +@prefix vendor: . + +ex:sequence a sbol:Sequence ; + sbol:hasNamespace ; + sbol:displayId "sequence" ; + sbol:elements "ACGTACGT" ; + sbol:encoding . + +ex:design a sbol:Component ; + sbol:hasNamespace ; + sbol:displayId "design" ; + sbol:type ; + sbol:hasSequence ex:sequence ; + sbol:hasFeature ; + vendor:catalog [ vendor:accession "external-catalog-record" ; + vendor:label "Descriptive label"@en ] . + + a sbol:SequenceFeature ; + sbol:displayId "site" ; + sbol:hasLocation . + + a sbol:Cut ; + sbol:displayId "cut" ; + sbol:hasSequence ex:sequence ; + sbol:at 4 . diff --git a/tests/provenance/test_activity.py b/tests/provenance/test_activity.py new file mode 100644 index 0000000..db45058 --- /dev/null +++ b/tests/provenance/test_activity.py @@ -0,0 +1,149 @@ +from datetime import UTC, datetime, timedelta + +import pytest +from rdflib import RDF, Graph, URIRef + +from lab.provenance import ( + Activity, + Agent, + AgentKind, + Association, + Component, + Document, + EvidenceState, + Implementation, + Plan, + Ref, + Usage, +) +from lab.provenance.vocabulary import DNA, LAB, PROV, SBOL + +NS = "https://example.org/provenance" + + +def test_qualified_provenance_has_correct_directions_and_owned_cardinalities(): + document = Document(namespace=NS) + design = Component(identity=document.iri("design"), types=(DNA,)) + software = Agent( + identity=document.iri("compiler"), kind=AgentKind.SOFTWARE, software_version="1.2" + ) + plan = Plan(identity=document.iri("method"), protocol=document.iri("protocol")) + planned = Activity( + identity=document.iri("planned"), + evidence_state=EvidenceState.PLANNED, + types=(LAB + "assembly",), + usage=(Usage(entity=design.ref, roles=(LAB + "design",)),), + association=(Association(agent=software.ref, plan=plan.ref, roles=(LAB + "planner",)),), + ) + output = Implementation( + identity=document.iri("output"), + derived_from=(design.ref,), + generated_by=(planned.ref,), + evidence_state=EvidenceState.PLANNED, + ) + document.add(design, software, plan, planned, output) + frozen = document.freeze() + graph = Graph().parse(data=frozen.to_turtle(), format="turtle") + assert ( + URIRef(output.identity), + URIRef(PROV + "wasGeneratedBy"), + URIRef(planned.identity), + ) in graph + assert (URIRef(planned.identity), URIRef(SBOL + "type"), URIRef(LAB + "assembly")) in graph + assert (URIRef(planned.identity), RDF.type, URIRef(LAB + "assembly")) not in graph + assert not list(graph.objects(URIRef(output.identity), URIRef(SBOL + "built"))) + native = frozen.to_sbol3() + assert len(native.objects) == 5 + activity = native.find(planned.identity) + assert len(activity.usage) == len(activity.association) == 1 + assert activity.association[0].agent == software.identity + assert activity.association[0].plan == plan.identity + assert activity.start_time is None and activity.end_time is None + assert not native.validate().errors + assert Document.from_sbol3(native).freeze() == frozen + + +def test_shared_predecessors_round_trip_without_reparenting(): + document = Document(namespace=NS) + parent = Activity(identity=document.iri("parent")) + left = Activity(identity=document.iri("left"), informed_by=(parent.ref,)) + right = Activity(identity=document.iri("right"), informed_by=(parent.ref,)) + document.add(parent, left, right) + native = document.to_sbol3() + assert len(native.objects) == 3 + assert native.find(left.identity).informed_by[0] is native.find(parent.identity) + assert native.find(right.identity).informed_by[0] is native.find(parent.identity) + assert Document.from_sbol3(native).freeze() == document.freeze() + + +def test_execution_timestamps_retain_timezone_and_evidence_kind(): + start = datetime(2026, 9, 27, 10, tzinfo=UTC) + document = Document(namespace=NS) + activity = Activity( + identity=document.iri("execution"), + start_time=start, + end_time=start + timedelta(seconds=5), + evidence_state=EvidenceState.SIMULATED, + ) + document.add(activity) + frozen = document.freeze() + restored = Document.from_sbol3(frozen.to_sbol3()).freeze() + assert restored == frozen + assert restored.resolve(activity.ref).evidence_state is EvidenceState.SIMULATED + + +@pytest.mark.parametrize( + "activity,code", + [ + (Activity(identity=NS + "/a", start_time=datetime(2026, 1, 1)), "timezone"), + ( + Activity( + identity=NS + "/a", + start_time=datetime(2026, 1, 2, tzinfo=UTC), + end_time=datetime(2026, 1, 1, tzinfo=UTC), + ), + "time-order", + ), + ( + Activity( + identity=NS + "/a", + start_time=datetime(2026, 1, 1, tzinfo=UTC), + evidence_state=EvidenceState.PLANNED, + ), + "planned-execution", + ), + ], +) +def test_invalid_execution_assertions_are_rejected(activity, code): + document = Document(namespace=NS) + document.add(activity) + with pytest.raises(ValueError, match=code): + document.freeze() + + +def test_planned_implementation_does_not_assert_realized_structure(): + document = Document(namespace=NS) + component = Component(identity=document.iri("design"), types=(DNA,)) + document.add( + component, + Implementation( + identity=document.iri("output"), + built=component.ref, + evidence_state=EvidenceState.PLANNED, + ), + ) + with pytest.raises(ValueError, match="planned-realization"): + document.freeze() + + +def test_activity_cycles_and_wrong_reference_types_are_rejected(): + document = Document(namespace=NS) + a = Activity(identity=document.iri("a"), informed_by=(Ref(document.iri("b")),)) + b = Activity(identity=document.iri("b"), informed_by=(a.ref,)) + document.add(a, b) + with pytest.raises(ValueError, match="dependency-cycle"): + document.freeze() + document = Document(namespace=NS) + document.add(Agent(identity=document.iri("b")), a) + with pytest.raises(ValueError, match="reference-type"): + document.freeze() diff --git a/tests/provenance/test_design.py b/tests/provenance/test_design.py new file mode 100644 index 0000000..a9e88df --- /dev/null +++ b/tests/provenance/test_design.py @@ -0,0 +1,216 @@ +from dataclasses import replace + +import pytest + +from lab.provenance import ( + Attachment, + Collection, + CombinatorialDerivation, + Component, + ComponentReference, + Constraint, + Cut, + Document, + EntireSequence, + Experiment, + ExperimentalData, + ExternallyDefined, + Implementation, + Interaction, + Interface, + LocalSubComponent, + Measure, + Model, + Participation, + Range, + Ref, + Sequence, + SequenceFeature, + SubComponent, + VariableFeature, +) +from lab.provenance.vocabulary import DNA, INLINE, IUPAC_DNA, OM, PRECEDES, SBOL + +NS = "https://example.org/design" + + +def test_structured_design_locations_interactions_and_derivations_round_trip(): + document = Document(namespace=NS) + sequence = Sequence(identity=document.iri("sequence"), elements="ACGTACGT", encoding=IUPAC_DNA) + inner_feature = LocalSubComponent(identity=document.iri("part/feature"), types=(DNA,)) + part = Component(identity=document.iri("part"), types=(DNA,), features=(inner_feature,)) + sub = SubComponent( + identity=document.iri("design/sub"), + instance_of=part.ref, + role_integration=SBOL + "mergeRoles", + roles=(NS + "/role",), + locations=(Range(sequence=sequence.ref, start=1, end=4, orientation=INLINE, order=1),), + source_locations=(EntireSequence(sequence=sequence.ref),), + ) + feature = SequenceFeature( + identity=document.iri("design/feature"), locations=(Cut(sequence=sequence.ref, at=4),) + ) + reference = ComponentReference(in_child_of=sub.ref, refers_to=inner_feature.ref) + external = ExternallyDefined(types=(DNA,), definition="https://example.org/catalog/item") + model = Model( + identity=document.iri("model"), + source="https://example.org/model.xml", + language="https://identifiers.org/edam:format_2585", + framework=NS + "/framework", + ) + design = Component( + identity=document.iri("design"), + types=(DNA,), + sequences=(sequence.ref,), + features=(sub, feature, reference, external), + models=(model.ref,), + constraints=(Constraint(restriction=PRECEDES, subject=sub.ref, object=feature.ref),), + interactions=( + Interaction( + types=(NS + "/interaction",), + participations=(Participation(roles=(NS + "/participant",), participant=sub.ref),), + ), + ), + interface=Interface(inputs=(sub.ref,), outputs=(feature.ref,)), + measures=(Measure(value=4.0, unit=OM + "nanogram"),), + ) + collection = Collection(identity=document.iri("collection"), members=(part.ref,)) + variants = CombinatorialDerivation( + identity=document.iri("variants"), + template=design.ref, + strategy=SBOL + "enumerate", + variable_features=( + VariableFeature( + cardinality=SBOL + "one", + variable=sub.ref, + variants=(part.ref,), + variant_collections=(collection.ref,), + variant_measures=(Measure(value=2.0, unit=OM + "one"),), + ), + ), + ) + outer_variants = CombinatorialDerivation( + identity=document.iri("outer_variants"), + template=design.ref, + variable_features=( + VariableFeature( + cardinality=SBOL + "one", variable=sub.ref, variant_derivations=(variants.ref,) + ), + ), + ) + attachment = Attachment( + identity=document.iri("attachment"), + source="https://example.org/data.csv", + format="https://identifiers.org/edam:format_3752", + size=4, + hash="abcd", + hash_algorithm="sha256", + ) + data = ExperimentalData(identity=document.iri("data"), attachments=(attachment.ref,)) + experiment = Experiment(identity=document.iri("experiment"), members=(data.ref,)) + implementation = Implementation(identity=document.iri("implementation"), built=design.ref) + document.add( + sequence, + part, + design, + model, + collection, + variants, + outer_variants, + attachment, + data, + experiment, + implementation, + ) + frozen = document.freeze() + native = frozen.to_sbol3() + assert not native.validate().errors + assert Document.from_sbol3(native).freeze() == frozen + assert Document.from_turtle(frozen.to_turtle()).freeze() == frozen + native_sub = native.find(sub.identity) + assert native_sub.role_integration == SBOL + "mergeRoles" + assert list(native_sub.roles) == [NS + "/role"] + assert native.find(feature.identity).locations[0].at == 4 + + +@pytest.mark.parametrize( + "location", + [ + Range(sequence=Ref(NS + "/sequence"), start=0, end=2), + Range(sequence=Ref(NS + "/sequence"), start=3, end=2), + Range(sequence=Ref(NS + "/sequence"), start=1, end=5), + Cut(sequence=Ref(NS + "/sequence"), at=-1), + Cut(sequence=Ref(NS + "/sequence"), at=5), + ], +) +def test_locations_are_checked_against_the_referenced_sequence(location): + document = Document(namespace=NS) + sequence = Sequence(identity=document.iri("sequence"), elements="ACGT", encoding=IUPAC_DNA) + document.add( + sequence, + Component( + identity=document.iri("design"), + types=(DNA,), + features=(SequenceFeature(locations=(location,)),), + ), + ) + with pytest.raises(ValueError, match="sequence-(range|cut)"): + document.freeze() + + +def test_constraint_references_must_belong_to_the_containing_component(): + document = Document(namespace=NS) + first = LocalSubComponent(identity=document.iri("first/feature"), types=(DNA,)) + other = LocalSubComponent(identity=document.iri("other/feature"), types=(DNA,)) + document.add( + Component(identity=document.iri("other"), types=(DNA,), features=(other,)), + Component( + identity=document.iri("first"), + types=(DNA,), + features=(first,), + constraints=(Constraint(restriction=PRECEDES, subject=first.ref, object=other.ref),), + ), + ) + with pytest.raises(ValueError, match="feature-scope"): + document.freeze() + + +def test_one_owned_object_cannot_belong_to_two_components(): + document = Document(namespace=NS) + feature = LocalSubComponent(identity=document.iri("shared"), types=(DNA,)) + first = Component(identity=document.iri("first"), types=(DNA,), features=(feature,)) + document.add(first, replace(first, identity=document.iri("second"))) + with pytest.raises(ValueError, match="duplicate-identity"): + document.freeze() + + +def test_component_containment_cannot_be_recursive(): + document = Document(namespace=NS) + document.add( + Component( + identity=document.iri("design"), + types=(DNA,), + features=(SubComponent(instance_of=Ref(document.iri("design"))),), + ) + ) + with pytest.raises(ValueError, match="dependency-cycle"): + document.freeze() + + +@pytest.mark.parametrize( + "obj", + [ + Component(identity=NS + "/a", types=()), + Component(identity=NS + "/a", types=(DNA, DNA)), + Sequence(identity=NS + "/a", elements=123, encoding=IUPAC_DNA), + Component(identity=NS + "/a", types=(DNA,), features=("invalid",)), + Component( + identity=NS + "/a", + types=(DNA,), + measures=(Measure(value=float("nan"), unit=OM + "one"),), + ), + ], +) +def test_invalid_model_field_values_fail_before_serialization(obj): + with pytest.raises(ValueError): + Document(namespace=NS).add(obj) diff --git a/tests/provenance/test_document.py b/tests/provenance/test_document.py new file mode 100644 index 0000000..7ca7940 --- /dev/null +++ b/tests/provenance/test_document.py @@ -0,0 +1,148 @@ +from dataclasses import FrozenInstanceError, replace +from typing import get_type_hints + +import pytest +import sbol3 + +import lab.provenance as provenance +from lab.provenance import Activity, Component, Document, Ref, Sequence, Usage +from lab.provenance.vocabulary import DNA, IUPAC_DNA + +NS = "https://example.org/provenance" + + +def test_freezing_assigns_owned_ids_without_mutating_authored_objects(): + document = Document(namespace=NS) + component = Component(identity=document.iri("design"), types=(DNA,)) + activity = Activity(identity=document.iri("build"), usage=(Usage(entity=component.ref),)) + document.add(activity, component) + frozen = document.freeze() + owned = frozen.resolve(activity.ref).usage[0] + assert owned.identity == NS + "/build/Usage1" + assert activity.usage[0].identity is None + with pytest.raises(ValueError, match="Assign an identity"): + _ = activity.usage[0].ref + assert frozen.resolve(owned.ref) == owned + with pytest.raises(FrozenInstanceError): + owned.name = "changed" + assert document.freeze() == frozen + document.add(Activity(identity=document.iri("later"))) + assert len(frozen.objects) == 2 + + +def test_owned_identity_allocation_preserves_explicit_ids_and_skips_collisions(): + document = Document(namespace=NS) + entity = Component(identity=document.iri("design"), types=(DNA,)) + activity = Activity( + identity=document.iri("build"), + usage=( + Usage(entity=entity.ref), + Usage(identity=document.iri("build/Usage1"), entity=entity.ref), + ), + ) + document.add(entity, activity) + assert [usage.identity for usage in document.freeze().resolve(activity.ref).usage] == [ + NS + "/build/Usage1", + NS + "/build/Usage2", + ] + + +def test_add_is_atomic_and_conflicting_definitions_are_rejected(): + document = Document(namespace=NS) + design = Component(identity=document.iri("design"), types=(DNA,)) + document.add(design, design) + with pytest.raises(ValueError, match="Conflicting definition"): + document.add(Activity(identity=document.iri("unused")), replace(design, name="different")) + assert document.objects == (design,) + with pytest.raises(ValueError, match="TopLevel"): + document.add(Usage(entity=design.ref)) + + +def test_reference_lookup_and_closure_are_explicit(): + document = Document(namespace=NS) + sequence = Sequence(identity=document.iri("sequence"), elements="ACGT", encoding=IUPAC_DNA) + design = Component(identity=document.iri("design"), types=(DNA,), sequences=(sequence.ref,)) + document.add(design) + assert not document.validate().is_valid + with pytest.raises(ValueError, match="unresolved-reference"): + document.freeze() + external = document.freeze(allow_external=True) + with pytest.raises(KeyError): + external.resolve(sequence.ref) + document.add(sequence) + frozen = document.freeze() + assert frozen.get(sequence.identity, Sequence) == frozen.resolve(sequence.ref) + with pytest.raises(TypeError, match="not Sequence"): + frozen.get(design.identity, Sequence) + assert Document.from_snapshot(frozen).freeze() == frozen + + +def test_no_process_global_namespace_and_native_documents_are_detached(): + previous = sbol3.get_namespace() + document = Document(namespace=NS) + design = Component(identity=document.iri("design"), types=(DNA,)) + document.add(design) + frozen = document.freeze() + native = frozen.to_sbol3() + native.find(design.identity).name = "Modified elsewhere" + assert frozen.resolve(design.ref).name is None + assert document.to_sbol3().find(design.identity).name is None + assert sbol3.get_namespace() == previous + + +def test_digest_is_independent_of_object_insertion_and_rdf_statement_order(): + first, second = Document(namespace=NS), Document(namespace=NS) + a = Activity(identity=first.iri("a")) + b = Activity(identity=first.iri("b")) + first.add(a, b) + second.add(b, a) + assert first.freeze().digest == second.freeze().digest + reversed_rdf = "\n".join(reversed(first.freeze().to_turtle().splitlines())) + assert Document.from_turtle(reversed_rdf).freeze().digest == first.freeze().digest + + +def test_local_io_is_reproducible_and_refuses_accidental_replacement(tmp_path): + document = Document(namespace=NS) + document.add(Activity(identity=document.iri("a"))) + path = tmp_path / "nested/provenance.ttl" + document.write(path) + document.write(path) + restored = Document.read(path) + assert restored.freeze() == document.freeze() + document.add(Activity(identity=document.iri("b"))) + with pytest.raises(FileExistsError): + document.write(path) + assert Document.read(path).freeze() == restored.freeze() + + +@pytest.mark.parametrize("value", ["relative", "", "https:///path", "https://example.org/bad name"]) +def test_refs_require_absolute_iris(value): + with pytest.raises(ValueError, match="absolute IRI"): + Ref(value) + + +@pytest.mark.parametrize("value", ["bad-name", "../escape", "123", "a//b", "", "a#b"]) +def test_document_iris_are_explicit_sbol_display_ids(value): + with pytest.raises(ValueError, match="display IDs"): + Document(namespace=NS).iri(value) + + +def test_every_public_model_has_runtime_resolvable_annotations(): + for name in provenance.__all__: + cls = getattr(provenance, name) + if isinstance(cls, type) and hasattr(cls, "__dataclass_fields__"): + get_type_hints(cls) + + +def test_mutable_collections_are_rejected_at_construction(): + with pytest.raises(TypeError, match="immutable"): + Component(identity=NS + "/design", types=[DNA]) + + +def test_mixed_namespaces_need_an_explicit_authoring_namespace_on_import(): + document = Document(namespace=NS) + document.add(Activity(identity=NS + "/a"), Activity(identity="https://elsewhere.org/b")) + rdf = document.freeze().to_turtle() + with pytest.raises(ValueError, match="multiple namespaces"): + Document.from_turtle(rdf) + assert Document.from_turtle(rdf, namespace=NS).freeze() == document.freeze() diff --git a/tests/provenance/test_sbol3.py b/tests/provenance/test_sbol3.py new file mode 100644 index 0000000..71599f4 --- /dev/null +++ b/tests/provenance/test_sbol3.py @@ -0,0 +1,127 @@ +from pathlib import Path + +import pytest +import sbol3 +from rdflib import RDF, Graph, Literal, URIRef +from rdflib.compare import isomorphic + +from lab.provenance import Activity, Component, Document, EvidenceState +from lab.provenance.vocabulary import SBOL + +NS = "https://example.org/native" +FIXTURE = Path(__file__).parents[1] / "fixtures/provenance/design.ttl" + + +def test_independently_authored_rdf_and_foreign_annotations_are_preserved(): + original = Graph().parse(FIXTURE, format="turtle") + document = Document.read(FIXTURE) + exported = Graph().parse(data=document.freeze().to_turtle(), format="turtle") + assert isomorphic(original, exported) + restored = Document.from_turtle(document.freeze().to_turtle()).freeze() + assert restored.digest == document.freeze().digest + native = restored.to_sbol3() + assert native.find("https://example.org/fixture/design/site/cut").at == 4 + assert isomorphic(original, native.graph()) + + +def test_native_provenance_document_import_has_no_inferred_execution_claims(): + native = sbol3.Document() + agent = sbol3.Agent(NS + "/agent") + plan = sbol3.Plan(NS + "/plan") + design = sbol3.Component(NS + "/design", [sbol3.SBO_DNA]) + activity = sbol3.Activity( + NS + "/activity", + usage=[sbol3.Usage(design.identity)], + association=[sbol3.Association(agent=agent, plan=plan)], + ) + native.add([agent, plan, design, activity]) + imported = Document.from_sbol3(native).freeze() + result = imported.get(activity.identity, Activity) + assert result.evidence_state is EvidenceState.UNKNOWN + assert result.start_time is None and result.end_time is None + assert result.usage[0].identity == activity.usage[0].identity + assert isomorphic(native.graph(), imported.to_sbol3().graph()) + + +def test_native_cut_and_role_integration_defects_are_normalized_only_at_adapter_boundary(): + native = sbol3.Document() + sequence = sbol3.Sequence(NS + "/sequence", elements="ACGT", encoding=sbol3.IUPAC_DNA_ENCODING) + part = sbol3.Component(NS + "/part", [sbol3.SBO_DNA]) + feature = sbol3.SubComponent( + part, role_integration=SBOL + "mergeRoles", locations=[sbol3.Cut(sequence, 2)] + ) + design = sbol3.Component(NS + "/design", [sbol3.SBO_DNA], features=[feature]) + native.add([sequence, part, design]) + imported = Document.from_sbol3(native).freeze() + sub = imported.get(design.identity, Component).features[0] + assert sub.role_integration == SBOL + "mergeRoles" + assert sub.roles == () + assert sub.locations[0].at == 2 + graph = Graph().parse(data=imported.to_turtle(), format="turtle") + assert ( + URIRef(feature.identity), + URIRef(SBOL + "roleIntegration"), + URIRef(SBOL + "mergeRoles"), + ) in graph + assert not list(graph.triples((None, URIRef(SBOL + "start"), None))) + # The source document remains untouched. + assert feature.role_integration == SBOL + "mergeRoles" + assert list(native.graph().triples((None, URIRef(SBOL + "start"), None))) + + +def test_input_with_multiple_scalar_values_is_not_silently_truncated(): + graph = Graph().parse(FIXTURE, format="turtle") + graph.add( + (URIRef("https://example.org/fixture/sequence"), URIRef(SBOL + "elements"), Literal("AAAA")) + ) + with pytest.raises(ValueError, match="at most one"): + Document.from_turtle(graph.serialize(format="turtle")) + + +@pytest.mark.parametrize("rdf_type", ["http://sbols.org/v2#ComponentDefinition", SBOL + "Unknown"]) +def test_unsupported_sbol_versions_and_classes_fail_explicitly(rdf_type): + text = f"<{NS}/object> <{RDF.type}> <{rdf_type}> ." + with pytest.raises(ValueError, match="SBOL2|Unsupported SBOL"): + Document.from_turtle(text, namespace=NS) + + +def test_unknown_sbol_properties_are_not_disguised_as_annotations(): + graph = Graph().parse(FIXTURE, format="turtle") + graph.add( + ( + URIRef("https://example.org/fixture/design"), + URIRef(SBOL + "unimplemented"), + Literal("value"), + ) + ) + with pytest.raises(ValueError, match="Unsupported SBOL property"): + Document.from_turtle(graph.serialize(format="turtle")) + + +def test_relative_iris_do_not_depend_on_the_working_directory(): + with pytest.raises(ValueError, match="relative"): + Document.from_turtle(f" <{RDF.type}> <{SBOL}Component> .", namespace=NS) + + +def test_display_ids_cannot_conflict_with_object_identity(): + graph = Graph().parse(FIXTURE, format="turtle") + graph.set( + (URIRef("https://example.org/fixture/design"), URIRef(SBOL + "displayId"), Literal("other")) + ) + with pytest.raises(ValueError, match="Invalid displayId"): + Document.from_turtle(graph.serialize(format="turtle")) + + +def test_orphaned_and_multiply_owned_children_are_rejected_on_import(): + graph = Graph().parse(FIXTURE, format="turtle") + graph.remove((URIRef("https://example.org/fixture/design"), URIRef(SBOL + "hasFeature"), None)) + with pytest.raises(ValueError, match="Orphaned"): + Document.from_turtle(graph.serialize(format="turtle")) + graph = Graph().parse(FIXTURE, format="turtle") + first = URIRef("https://example.org/fixture/design") + second = URIRef("https://example.org/fixture/second") + for _, predicate, obj in tuple(graph.triples((first, None, None))): + if predicate != URIRef(SBOL + "displayId"): + graph.add((second, predicate, obj)) + with pytest.raises(ValueError, match="more than one owner"): + Document.from_turtle(graph.serialize(format="turtle")) diff --git a/uv.lock b/uv.lock index 7772733..1cdfdd6 100644 --- a/uv.lock +++ b/uv.lock @@ -235,6 +235,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fe/5e/3be305568fe5f34448807976dc82fc151d76c3e0e03958f34770286278c1/flexparser-0.4-py3-none-any.whl", hash = "sha256:3738b456192dcb3e15620f324c447721023c0293f6af9955b481e91d00179846", size = 27625 }, ] +[[package]] +name = "html5lib" +version = "1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, + { name = "webencodings" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ac/b6/b55c3f49042f1df3dcd422b7f224f939892ee94f22abcf503a9b7339eaf2/html5lib-1.1.tar.gz", hash = "sha256:b2e5b40261e20f354d198eae92afc10d750afb487ed5e50f9c4eaf07c184146f", size = 272215 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6c/dd/a834df6482147d48e225a49515aabc28974ad5a4ca3215c18a882565b028/html5lib-1.1-py2.py3-none-any.whl", hash = "sha256:0d78f8fde1c230e99fe37986a60526d7049ed4bf8a9fadbad5f00e22e58e041d", size = 112173 }, +] + [[package]] name = "id" version = "1.6.1" @@ -277,6 +290,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484 }, ] +[[package]] +name = "isodate" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/db/7a/c0a56c7d56c7fa723988f122fa1f1ccf8c5c4ccc48efad0d214b49e5b1af/isodate-0.6.1.tar.gz", hash = "sha256:48c5881de7e8b0a0d648cb024c8062dc84e7b840ed81e864c7614fd3c127bde9", size = 28443 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b6/85/7882d311924cbcfc70b1890780763e36ff0b140c7e51c110fc59a532f087/isodate-0.6.1-py2.py3-none-any.whl", hash = "sha256:0751eece944162659049d35f4f549ed815792b38793f07cf73381c1c87cbed96", size = 41722 }, +] + [[package]] name = "jaraco-classes" version = "3.4.0" @@ -359,6 +384,8 @@ source = { editable = "." } dependencies = [ { name = "pint", version = "0.25.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "pint", version = "0.26.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, + { name = "rdflib" }, + { name = "sbol3" }, ] [package.optional-dependencies] @@ -384,6 +411,8 @@ requires-dist = [ { name = "opentrons", marker = "extra == 'opentrons'", specifier = "==8.8.2" }, { name = "pint", specifier = ">=0.24,<0.27" }, { name = "pylabrobot", marker = "extra == 'star'", specifier = "==0.2.1" }, + { name = "rdflib", specifier = ">=6.1.1,<7" }, + { name = "sbol3", specifier = "==1.2.0.post0" }, ] provides-extras = ["opentrons", "star"] @@ -588,6 +617,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/98/40/a6620ced846254d020e350cccdf19cd505628e8dca1b82b93cfcddb242f8/opentrons_shared_data-8.8.2-py3-none-any.whl", hash = "sha256:12d77eab3a41bfb4e5a947f7ca5062d40eb1fcffcf2a6d12a317af7ac249b6f3", size = 1169425 }, ] +[[package]] +name = "owlrl" +version = "6.0.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "rdflib" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7d/c7/208aece36279e4f1236e437119358786e39530ecc1719d4e1afeddba5288/owlrl-6.0.2.tar.gz", hash = "sha256:904e3310ff4df15101475776693d2427d1f8244ee9a6a9f9e13c3c57fae90b74", size = 57179 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9d/56/11fe63c2c317347f69be17e9ece1991e0ec6c2cdb8225c0baa5b96e283ed/owlrl-6.0.2-py3-none-any.whl", hash = "sha256:57eca06b221edbbc682376c8d42e2ddffc99f61e82c0da02e26735592f08bacc", size = 54516 }, +] + [[package]] name = "packaging" version = "26.3" @@ -660,6 +701,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538 }, ] +[[package]] +name = "prettytable" +version = "3.18.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wcwidth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/81/74/ba08d81e668ccfe8658d7520a307e63c19862c08eb4ccb26f356c5239a7a/prettytable-3.18.0.tar.gz", hash = "sha256:439217116152244369caf3d9f1caf2f9fe29b03bd79e88d2928c8e718c95d680", size = 76373 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/be/2e6798ace5cc036f5d05d36b7b2fd85346f1a708c87060890b070d0ec607/prettytable-3.18.0-py3-none-any.whl", hash = "sha256:b3346e0e6f79180833aebaac088ae926340586cf6d7d991b9eb125b65f72313a", size = 37357 }, +] + [[package]] name = "pycparser" version = "3.0" @@ -777,6 +830,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/c0/4a1d8d6d6ec6d4fc1c1f25257c3a71be07b438698fb24eedf7c14bc14313/pylabrobot-0.2.1-py3-none-any.whl", hash = "sha256:8b7b58d1c31a7b19b9d6a11f10ba1303c6222b6a582a8a95f18c6abcf41408cd", size = 1174640 }, ] +[[package]] +name = "pyparsing" +version = "3.3.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e4/11/b213bebff182584360cb8d17c72c1677fec5c5c228de439e63bcf8ab1c8f/pyparsing-3.3.3.tar.gz", hash = "sha256:928ae7e20211f3b6f3915a72f06a0cfd29ab9d24279dd6346b6b1a7146397d36", size = 1050487 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/38/bb/d215ee7c73b61497b28a5503f9f53523f294fcc936762b7caf90e0c1c2b5/pyparsing-3.3.3-py3-none-any.whl", hash = "sha256:ece8c00a69cf01b45d0b1dedabb469c90d8caf996d4fda40f147627a122849a4", size = 126420 }, +] + [[package]] name = "pyrsistent" version = "0.20.0" @@ -807,6 +869,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/07/bc/587a445451b253b285629263eb51c2d8e9bcea4fc97826266d186f96f558/pyserial-3.5-py2.py3-none-any.whl", hash = "sha256:c4451db6ba391ca6ca299fb3ec7bae67a5c55dde170964c7a14ceefec02f2cf0", size = 90585 }, ] +[[package]] +name = "pyshacl" +version = "0.28.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "importlib-metadata", marker = "python_full_version < '3.12'" }, + { name = "owlrl" }, + { name = "packaging" }, + { name = "prettytable" }, + { name = "rdflib", extra = ["html"] }, +] +sdist = { url = "https://files.pythonhosted.org/packages/61/55/370d2bae6eb2d6302caac1eba2c1ad365fbb43c4d4805d00367839b3089c/pyshacl-0.28.1.tar.gz", hash = "sha256:c98e1927541f0cbeb0ba27f3fbe585d638281d6e63f314462dcc32282b3c8353", size = 1399231 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/17/57f5e48a856b70715b86b228ed4211fa0447e38c32eec6fd7c30239d5ea9/pyshacl-0.28.1-py3-none-any.whl", hash = "sha256:345fe96fb2f38218c53a0f977329ed0aa1a54256526bf6de62243ac7ae823d6b", size = 1321044 }, +] + [[package]] name = "pytest" version = "9.1.1" @@ -836,6 +914,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930 }, ] +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892 }, +] + [[package]] name = "python-dotenv" version = "1.2.3" @@ -863,6 +953,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/de/3d/8161f7711c017e01ac9f008dfddd9410dff3674334c233bde66e7ba65bbf/pywin32_ctypes-0.2.3-py3-none-any.whl", hash = "sha256:8a1513379d709975552d202d942d9837758905c8d01eb82b8bcc30918929e7b8", size = 30756 }, ] +[[package]] +name = "rdflib" +version = "6.3.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "isodate" }, + { name = "pyparsing" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c8/28/4d1f27c5d73f58e567ca1a14a4eab7d7978a09c4e117687f9f6c216d3366/rdflib-6.3.2.tar.gz", hash = "sha256:72af591ff704f4caacea7ecc0c5a9056b8553e0489dd4f35a9bc52dbd41522e0", size = 4749592 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/af/92/d7fb1d7fb70c9f7003fa50b7a3880ebcb311cc3f8552b3595e7c8f75aeeb/rdflib-6.3.2-py3-none-any.whl", hash = "sha256:36b4e74a32aa1e4fa7b8719876fb192f19ecd45ff932ea5ebbd2e417a0247e63", size = 528122 }, +] + +[package.optional-dependencies] +html = [ + { name = "html5lib" }, +] + [[package]] name = "readme-renderer" version = "46.0" @@ -951,6 +1059,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fe/a0/50787329e4f20bf9dc9f6230015d46ec69c51a97ace5bc202dae4755365d/ruff-0.16.8-py3-none-win_arm64.whl", hash = "sha256:d075e820af612102ce217f07cc93e69f9490b10ec13ea85fa87bd03d996cef8a", size = 10386316 }, ] +[[package]] +name = "sbol3" +version = "1.2.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyshacl" }, + { name = "python-dateutil" }, + { name = "rdflib" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1b/b3/24fc213e588cb2bfc0ec2f99b1ec8366bdbe42ecb096528cb2cacad5eb5e/sbol3-1.2.0.post0.tar.gz", hash = "sha256:e0ddeaf44a29495da043287667275a663629862820acf5456bc71899c940dd8b", size = 86027 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/77/43/6b787610144ba23da2577549d3be0e74351767c84d514eb34232f473124a/sbol3-1.2.0.post0-py3-none-any.whl", hash = "sha256:742188cacff5829816c61a2f0e82c15357c01821362f66e863530b0fac1b442f", size = 73510 }, +] + [[package]] name = "secretstorage" version = "3.5.0" @@ -964,6 +1086,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b7/46/f5af3402b579fd5e11573ce652019a67074317e18c1935cc0b4ba9b35552/secretstorage-3.5.0-py3-none-any.whl", hash = "sha256:0ce65888c0725fcb2c5bc0fdb8e5438eece02c523557ea40ce0703c266248137", size = 15554 }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050 }, +] + [[package]] name = "twine" version = "6.2.0" @@ -1014,6 +1145,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717 }, ] +[[package]] +name = "wcwidth" +version = "0.9.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/dc/ac/3a943d2792c9bb368aaa8b50121c0f778460ba2d7fbdc0a0366201d9e761/wcwidth-0.9.1.tar.gz", hash = "sha256:5823209b0d43af322ce698c689380d7c15ca31fa8e6e3be8459f27031bef0af5", size = 952410 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/48/7c/130de33c0a7f6efecee28fef5711fc1f8d55ac33885a6b85f4f6689004ba/wcwidth-0.9.1-cp310-abi3-macosx_10_9_x86_64.whl", hash = "sha256:10b00ba23482e352f874d2e8135e7ace9da838646c7dd800566246bbd46125ff", size = 604219 }, + { url = "https://files.pythonhosted.org/packages/33/7c/c7bb03da54de7ac5a08753e186b963f91847b04ed8ae5be136a4d5f52b75/wcwidth-0.9.1-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:40d936d72c9bdc10df43f93a8be502bc5024b487259139f66a328722c07f34a9", size = 609038 }, + { url = "https://files.pythonhosted.org/packages/0b/71/f57effdf895c2d6d49333f1d6833a9f5438725c55a04ed3124a1c95bb3aa/wcwidth-0.9.1-cp310-abi3-manylinux1_i686.manylinux2014_i686.manylinux_2_17_i686.manylinux_2_5_i686.whl", hash = "sha256:69bb970cf5652b88cfdb9d3fdd1764fc15e5f8ad531643e7bb3e894cd969740e", size = 761464 }, + { url = "https://files.pythonhosted.org/packages/4f/16/15b03d8fb7a747d0d2cf4a33da7b1b3e092493c58780f6589d5b1e9e3b49/wcwidth-0.9.1-cp310-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:b5da43d6967668982e44a52fb551967d293f86d26cd86036ac95bbdd34394ed9", size = 770483 }, + { url = "https://files.pythonhosted.org/packages/e2/a0/834886b30e3b885a5a96e8358f1b3917bc0bac43ae4178d77f0c210d079b/wcwidth-0.9.1-cp310-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:747fb724223f417a17541a95a17c1dac3a8ef9a0cf41684950f0eab191a35f65", size = 768443 }, + { url = "https://files.pythonhosted.org/packages/a2/ae/1b7597e0132ebc2c920632f39d8c1c61881345c9b480c0a95eb91e7460fa/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:6e1272b7986cefe79783737e38bdb9eaae0682b333c7bd132024441193dd5ce7", size = 781171 }, + { url = "https://files.pythonhosted.org/packages/fb/bb/32149e3b481953f4ef35f38f166e2532886d33ae54ccdd71393f821dfad9/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_i686.whl", hash = "sha256:fe021c4d8de9d36c31a0cb41d0d2546dadd3b0708a301f1a0c66ce200851831f", size = 781576 }, + { url = "https://files.pythonhosted.org/packages/d6/72/1ee3dff67697bda93bc56ec9f7b3b89609cd64b9abb328512a2f22eb9e47/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:708158c082364af442f9983de7b6ec9ac0d2e1b825ada25f0911b1f138d55405", size = 783526 }, + { url = "https://files.pythonhosted.org/packages/ce/0a/e94f19a60f6127bcf3dfcf223ca05a37650d57b37b5ed623597c842ce7ca/wcwidth-0.9.1-cp310-abi3-win32.whl", hash = "sha256:356376852357b8fca71fe5415808ec421679e04b4a98eb7c9cb6a7984b911a05", size = 592136 }, + { url = "https://files.pythonhosted.org/packages/3c/ff/d884d2ec7dcdc86cda742a86c7160bc726dbccae76d7ae1f9ac833c26680/wcwidth-0.9.1-cp310-abi3-win_amd64.whl", hash = "sha256:991d1c8834f548e9c1f16432075ee84638e122312556bbf1ed595ea8fffc4673", size = 596738 }, + { url = "https://files.pythonhosted.org/packages/fa/93/16e30d617b937272a2310a6e3c1f9d8f94984e19ad4a96eb935628e3e6b6/wcwidth-0.9.1-cp310-abi3-win_arm64.whl", hash = "sha256:61bd7aef9cafb6cb77a37a169998d7928ce82a51522146d11f60db9e7d1cb43a", size = 594700 }, + { url = "https://files.pythonhosted.org/packages/96/a2/f06f2e0be4895e5943d358755b51968ace85774cf0a4cc42b4dafb793832/wcwidth-0.9.1-py3-none-any.whl", hash = "sha256:e0c3a1c45c5b9550c6919a4449e95f5b177f6e165786376981db8f1addae9b21", size = 300656 }, +] + +[[package]] +name = "webencodings" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d5/a0/8fd707bcb776a7be556bad06a2ea5fb9bd519df78ef8e26f70ccf0f38bff/webencodings-0.6.1.tar.gz", hash = "sha256:565f9ad031c702dae404e27a099e3e09186a3ab1b9520f06d215502b651fd910", size = 15001 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/77/c6/040cbc72480d789a5f40d63fb484d3106554c4dfa2d2b70ad5022057750f/webencodings-0.6.1-py3-none-any.whl", hash = "sha256:7fab6269c8bf237c657876b52058ccb182e861518d1c695c1a9aaa8c1c105d5b", size = 8745 }, +] + [[package]] name = "websockets" version = "17.1"