Runtime API reference

Runtime API reference#

These signatures and docstrings are imported from the installed plumb package during the book build. The Model lifecycle explains when to use each operation.

Model#

class plumb.Model(namespace, *, name=None, ontology_iri=None, imports=(), prefixes=None)#

A model under construction: a namespace plus the entities in it.

system(name, *, label=None)#: Collect equipment created in a scope into an S223 System.

by_class()#

Group current entities by their exact generated Python class.

Keys are classes, so they can be imported and passed to find_all. Values follow model insertion order. Deferred connections and ports appear only in a resolved snapshot, not in this authored inventory.

classmethod from_graph(source, *, registries=None, namespace=None, infer=False, shapes=None, strict=True)#

Hydrate a mutable object model backed by an RDF graph.

When registries is omitted, the loader selects installed ontology packages from the graph’s owl:imports declarations. Pass a registry explicitly for graphs without those declarations or for custom packages, for example plumb.watr.drinking_water.REGISTRY. When infer is true, shifty applies SHACL-AF rules before objects and their relationships are materialized; the rules come from shapes or, by default, the selected packages’ bundled shapes, as in validate(). Mutating generated attributes writes through to the graph; RDF outside the generated object projection is preserved. Unknown RDF classes remain in graph() but are not exposed as Python objects.

find_all(entity_type=None, /, *, limit=None, attrs=None, **attributes)#

Return objects of entity_type matching the given attributes.

Subclasses are included; pass a tuple for several types or exact(cls) for one class only. Filters match by equality, and a class or enum value also matches anything narrower; see docs/traversal.md for lists, regexes, presence tests, and callables. On multi-valued relationships a filter means membership, so find_all(AHU, feeds=vav) works as expected. attrs passes filters whose names collide with these keyword options.

find(entity_type=None, /, *, attrs=None, **attributes)#: Return the first matching object, or None.

export_ontology(destination=None)#

Return the portable schema graph for generated configurations in this model.

The extension ontology graph is stored with the generated registry and includes class identities, parent links, and imports. When destination is supplied it is serialized as Turtle and the graph is also returned.

check()#: Check deferred S223 constraints without materializing a snapshot.

resolve()#: Resolve all deferred intentions into a cached frozen snapshot.

compile(*, validate=True, shapes=None, infer=True)#

Resolve and, by default, validate one concrete immutable graph.

Validation uses the exact SHACL graph bundled with each generated package when available. Pass validate=False only for deliberate diagnostic or partial-model workflows.

graph()#: Resolve the authored model and return its concrete RDF graph.

save(path, *, format=None)#: Serialize to a file; format inferred from the extension (.ttl, .nt, .xml, .jsonld) unless given.

validate(shapes=None, *, infer=True)#

Validate against SHACL shapes (via shifty).

shapes may be an rdflib Graph, a path, or a URL; when omitted, the shapes graphs advertised by the generated packages used in this model are collected automatically (each package knows its ontology source). Violations are mapped back to entities where possible.

classmethod Entity.describe_ports()#

Summarize the class’s port directions and SHACL requirements.

This describes what the ontology permits or requires, rather than the concrete and deferred points on a particular instance. Use instance.ports to inspect those points.

Entity.ports#

Concrete connection points and authored deferred ports on this entity.

The authored model can contain both generated ConnectionPoint entities and PortHandle intentions. A resolved snapshot exposes concrete points here after the solver has materialized the deferred ones.

Traversal#

class plumb.core.Entity(name=None, *, label=None, comment=None, model=None, **props)

Base class for all generated ontology classes.

Construction requires an active Model (either ambient, via with Model(...), or passed as model=). The first argument is the entity’s local name; the model’s namespace turns it into an IRI at serialization time. When omitted, a name is auto-generated per model as <class>_<n> (vav_1, vav_2, …).

Reserved attribute names (name, type, label, comment, meta, model, ports, describe_ports, the find and walk methods, …; see plumb.gen.naming.RESERVED): ontology properties that would collide are renamed by the generator’s collision policy.

property contents: Entities this one directly contains, in model order.

property descendants: Everything this entity contains, depth-first, each entity once.

property container: The direct container; raises ModelingError if there are several.

property containers: Every container of this entity, nearest first.

find_all(entity_type=None, /, *, recursive=True, limit=None, attrs=None, **attributes)

Contained entities matching a type and filters, as Model.find_all.

Searches everything this entity contains, or only its direct contents with recursive=False.

find(entity_type=None, /, *, recursive=True, attrs=None, **attributes): The first contained entity matching a type and filters, or None.

find_parents(entity_type=None, /, *, limit=None, attrs=None, **attributes): Containers matching a type and filters, nearest first.

find_parent(entity_type=None, /, *, attrs=None, **attributes): The nearest container matching a type and filters, or None.

walk(direction='downstream', *, through=None, descend=False, include_connections=False, max_hops=None)

Walk the connection graph breadth-first, nearest entities first.

through prunes the path: a mapping of filters checked against each pipe or connection crossed, or a callable given each Crossing. descend steps into containers through maps_to. Pipes and connections are yielded only with include_connections. See docs/traversal.md.

find_all_next(entity_type=None, /, *, through=None, descend=False, include_connections=False, max_hops=None, limit=None, attrs=None, **attributes)

Downstream entities matching a type and filters, nearest first.

Filters apply to the results; through applies to the path.

find_next(entity_type=None, /, *, through=None, descend=False, include_connections=False, max_hops=None, attrs=None, **attributes): The nearest downstream entity matching a type and filters.

find_all_previous(entity_type=None, /, *, through=None, descend=False, include_connections=False, max_hops=None, limit=None, attrs=None, **attributes): Upstream entities matching a type and filters, nearest first.

find_previous(entity_type=None, /, *, through=None, descend=False, include_connections=False, max_hops=None, attrs=None, **attributes): The nearest upstream entity matching a type and filters.

plumb.exact(value)#: Match value itself: the exact class, enum value, or object.

class plumb.Hop(entity, depth, via)#: An entity reached by a walk, its distance, and the path taken.

class plumb.Crossing(exit, connection, entry)#

One step of a walk.

connection is the pipe or connection crossed, or None for a maps_to step into or out of a container. exit is None only for the first step of a walk that starts on a connection.

Property schemas#

class plumb.PropertySchema#

Base for frozen value schemas with explicit typed constructors.

Schemas are usually generated from SHACL. A subclass written in Python declares each field with quantity_field() and becomes a frozen, keyword-only dataclass without a decorator.

classmethod describe()#: Describe fields, model notation, units, and source shape provenance.

class plumb.qudt.QuantityAmount(value, unit)#: A model-independent numeric value with an explicit QUDT unit.

class plumb.PropertySet(properties)#: Read-only mapping from schema field names to newly attached entities.

Entity.apply_properties(schema, *, prefix=None)#: Attach fresh properties from a generated RDF/SHACL value schema.

Entity.quantity(name, value, *, unit=None, kind, capability='static', label=None, **properties)#

Create and attach a named S223 quantifiable property to this object.

capability selects a static, observable, or actuatable property. Measurement aspects are independent; pass has_aspect explicitly. A QuantityAmount supplies its own unit. The returned property is an ordinary model entity.

class plumb.PointRef(owner, name, port)#

Stable reference to one declared point on one equipment instance.

property medium#: Current medium hint for the referenced connection point.

property maps_to#: The container’s point this one stands for, as s223:mapsTo.

apply_properties(schema, *, prefix=None)#: Attach values to this exact authored or hydrated point.

quantity(name, value, *, kind, unit=None, capability='static', label=None, **properties)#: Create a scalar property on this exact authored or hydrated point.

class plumb.PointGroupRef(owner, name, points)#: Stable reference to a declared interchangeable point group.

class plumb.PortHandle(model, owner, name, direction, *, medium=None, label=None, fixed_medium=False)#

Stable authored name and constraints for a deferred connection point.

property direction#: The fixed direction or direction domain authored for this port.

pair(other)#: Pair two internal ports and constrain them to one medium.

property maps_to#

The container’s point this port stands for, as s223:mapsTo.

The target may be another port or a concrete point. Resolution gives both the same medium and writes the link onto the concrete points.

quantity(name, value, *, unit=None, kind, capability='static', label=None, **properties)#: Attach a quantity to the connection point created for this port.

apply_properties(schema, *, prefix=None)#: Attach a generated property schema to this port’s concrete point.

Class discovery#

Use SomeClass.meta["ChildName"] for a descendant scoped to that class. The lookup also accepts the full Python name or ontology class IRI.

class plumb.core.ClassMetaView(cls)#

Read-only view of a generated class’s ontology provenance.

Accessed as AHU.meta. Everything here comes from the ontology at generation time; the generated docstrings render the same data.

property iri#: IRI of the ontology class.

property label#: rdfs:label of the class.

property definition#: skos:definition / rdfs:comment text.

property ontology#: The ontology this class was generated from.

property registry#: The generated package’s class registry.

property parents#: Direct ontology superclasses (as Python classes).

property subclasses#: Direct subclasses registered in this generated package.

property children#: Direct child classes; alias for subclasses.

property descendants#: All registered subclasses at any depth, sorted by Python name.

tree(*, max_depth=None)#: Render this class and its registered subclass hierarchy as text.

property properties#: Effective property specs: inherited plus own, own winning.

property port_contract#

Inherited SHACL requirements for connection points on this class.

Slots describe unconditional count, direction, point class, and allowed medium roots. Boolean alternatives remain in constraints. Named shape slots are included where compiled; inspect instance.ports for the concrete and authored points on a particular instance.

describe_ports()#: Render this class’s inherited port requirements for authoring.

property abstract#: True for organizational classes not meant to be instantiated.

class plumb.core.Registry(ontology)#

Per-generated-package index of classes and UX negotiation tables.

Each generated package creates exactly one Registry; every class in the package registers itself here (via Entity.__init_subclass__) so property ranges — stored as strings in PropertySpec — can be resolved to real classes at runtime without import-order gymnastics.

resolve(name)#: Look up a class by its Python name (used for range checking).

property classes#: All currently registered classes, sorted by Python name.

subclasses(base, *, recursive=False)#: Registered subclasses of base, directly or at any depth.

tree(base, *, max_depth=None)#

Render the registered subclass tree rooted at base.

max_depth=0 shows only the root. With multiple inheritance, a class appears beneath each of its direct parents.

class plumb.core.EnumValue(iri, *, label=None, definition=None, parent=None)#

One enumeration member (a punned class/individual in the ontology).

is_a(other)#: True if this value is other or a descendant of it.

property children#: Direct enum children, sorted by ontology-local name.

property descendants#: All values below this one, in depth-first hierarchy order.

tree_text(*, max_depth=None, show_iris=False)#

Return a deterministic text tree rooted at this value.

max_depth=0 shows only this value (and an ellipsis when it has children); the default walks every descendant. Set show_iris to display full RDF identifiers instead of ontology-local names.

tree(*, max_depth=None, show_iris=False, file=None)#: Print the hierarchy rooted at this value.

class plumb.core.PortContract(slots, constraints)#

Inherited requirements for points a solver may place on an entity.

slots are unconditional requirements. constraints preserve and/or/xone alternatives from compiled SHACL shapes. Neither structure assigns names to points that have not been authored or resolved.

describe()#: Summarize directions, counts, media, and nested shape constraints.

class plumb.core.PortSlot(point_class, direction, medium, medium_options, min_count, max_count, name=None, label=None)#

One unconditional connection-point requirement, resolved for inspection.

medium and medium_options are allowed hierarchy roots, not necessarily exact media. Unregistered enum IRIs remain strings.

Connections and snapshots#

plumb.connect(a, b, *, medium=None, connection=None, name=None, ports='match')#

Connect a to b along the flow direction; the functional form of a >> b for when you need to retain the connection intention.

For S223/WaTr packages this returns a stable ConnectionHandle. Named and unnamed connections select ports by the same rules. Use ports="new" to allocate fresh directed ports on bare equipment, or pass explicit point references. Port identity is independent of the name. medium= gives the Z3 resolver a flow-media hint, and the handle’s medium property may be refined later. connection= constrains the materialized Connection subclass. Retrieve the concrete entity with resolved.connection(handle) after Model.resolve(). Generated multi-valued properties can be staged with collection syntax, for example handle.has_role.add(role). For Brick-style packages it adds the flow property edge (feeds) and returns None (there is no intermediate resource to return).

class plumb.ConnectionHandle(model, identifier, source, target, *, medium, connection, name)#

Stable, editable reference to a deferred connection intention.

Generated Connection properties are staged on the handle before resolution. Multi-valued properties expose a small set-like proxy, so handle.has_role.add(role) mirrors authored entity relationships.

property medium#

The authored medium constraint, or None if unconstrained.

Assignment invalidates cached resolution so the next check or resolve uses the new hint.

apply_properties(schema, *, prefix=None)#: Attach a generated schema to the eventual connection.

quantity(name, value, *, kind, unit=None, capability='static', label=None, **properties)#: Create a scalar property on the eventual connection.

assign(**properties)#: Stage properties, including custom-subclass escape-hatch values.

class plumb.ResolvedModel(model, connections, explanations, source_revision, authored_model=None)#

Frozen concrete model produced from authored entities and intentions.

find_all(entity_type=None, /, **options)#: Model.find_all over the concrete snapshot entities.

find(entity_type=None, /, **options)#: Model.find over the concrete snapshot entities.

connection(handle)#: Return the concrete connection corresponding to handle.

point(reference)#: Resolve a named point reference to its concrete snapshot point.

explain(handle, target=None)#

Explain a connection by handle or by its authored endpoints.

Passing both endpoints also works for links authored with >>, whose operator result does not expose the internal connection handle.

class plumb.ResolutionReport(status, issues=())#: Non-materializing result returned by Model.check().

class plumb.ResolutionExplanation(connection, medium, source_point, target_point, medium_reason, evidence=())#: Human-readable provenance for one materialized connection decision.

Validation and errors#

class plumb.ValidationReport(ok, issues)#

Outcome of Model.validate().

Truthy when there are no violations; warnings/infos (223 uses info severity for advisory notices like dangling boundary connection points) are listed in issues but don’t fail the model.

class plumb.core.Issue(severity, message, entity, focus, path)#

One validation result, mapped back to the authoring layer.

severity#: violation | warning | info.

message#: Human-readable description (SHACL resultMessage).

entity#: The entity the result focuses on, when it maps to one.

focus#: The raw focus node IRI (useful when entity is None).

path#: The offending property path (IRI), if reported.

exception plumb.AmbiguousModelError(message, *, alternatives=())#: Several semantically distinct concrete graphs remain possible.

exception plumb.UnsatisfiableModelError(message, *, core=())#: The authored constraints have no solution.

exception plumb.ValidationError(message, *, report=None)#: A resolved graph does not conform to its compiled SHACL shapes.