Connect equipment

Connect equipment#

connect(a, b) records that flow goes from a to b. You do not create pipes or connection points. When the model resolves, the solver picks the points, the medium, and the connection class. See How connections are resolved.

Read a class’s port requirements#

describe_ports() lists the connection points a class requires: direction, count, and allowed media.

from plumb import Model, connect
from plumb.watr import GravityThickener, Pump, enums
from plumb.watr.drinking_water import ClarificationStage, RapidSandFiltrationStage

values = enums.values
print(Pump.describe_ports())

Pump port requirements
Unconditional ports:
  - direction="in" -> InletConnectionPoint (count 1..any); medium roots: Constituent-Refrigerant, Mix-Fluid
  - direction="out" -> OutletConnectionPoint (count 1..any); medium roots: Constituent-Refrigerant, Mix-Fluid

"in" is flow arriving at the equipment, "out" is flow leaving it, and "bi" is either. A medium root admits everything beneath it. Domain classes also list their named points:

print(ClarificationStage.describe_ports())

ClarificationStage port requirements
Unconditional ports:
  - clarified_water: direction="out" -> OutletConnectionPoint (count 1..1); medium roots: Fluid-Water
  - inlet: direction="in" -> InletConnectionPoint (count 1..1); medium roots: Fluid-Water
  - sludge: direction="out" -> OutletConnectionPoint (count 1..1); medium roots: Fluid-Sludge
  - direction="out" -> OutletConnectionPoint (count 2..any); medium roots: Mix-Fluid

The same information is structured on meta.port_contract:

for slot in Pump.meta.port_contract.slots:
    print(slot.direction, slot.point_class.__name__, slot.min_count, slot.max_count)

in InletConnectionPoint 1 None
out OutletConnectionPoint 1 None

Connect and check#

model = Model("urn:example:residuals#")
with model:
    clarifier = ClarificationStage("clarifier")
    sand_filter = RapidSandFiltrationStage("sand-filter")
    sludge_pump = Pump("sludge-pump")
    thickener = GravityThickener("thickener")

sludge = connect(clarifier, sludge_pump, name="sludge-draw")
report = model.check()
print(report.status)
for issue in report.issues:
    print(issue.message)

underconstrained
connection or named point choices remain ambiguous: sludge-draw; fix a point direction or medium, or connect it to equipment that determines the choice

check() solves without building anything. It returns one of:

  • complete: exactly one answer.

  • underconstrained: more than one. Here the clarifier has a water and a sludge outlet, and a WaTr pump accepts either.

  • unsatisfiable: none. The issues name the authored facts in conflict.

An underconstrained report is truthy; test report.complete. resolve(), graph(), save(), and validate() raise AmbiguousModelError or UnsatisfiableModelError instead of guessing.

Add a hint#

Add the narrowest fact you know. Here, the medium:

sludge.medium = values.Fluid_Sludge
connect(sludge_pump, thickener, name="sludge-to-thickener")
print(model.check().status)

complete

The second connection needs no hint: the pump now carries sludge.

Hint

Example

Medium

connect(a, b, medium=...) or handle.medium = ...

Named point

connect(clarifier.clarified_water, ...)

Declared port

equipment.port("dosing", direction="in", medium=...)

Connection class

connect(a, b, connection=Pipe)

A named point selects that point outright:

connect(clarifier.clarified_water, sand_filter.inlet, name="clarified-water")
print(model.check().status)

complete

.port() declares a point the shapes do not mention, with a stable name. Its medium= overrides the class’s medium domain.

Tank(..., medium=[...]) records s223:hasMedium on the tank but does not constrain its connections. Put the medium on the port or connection instead.

Inspect the result#

resolved = model.resolve()
resolved.explain(sludge)

sludge-draw: clarifier-sludge -> sludge-pump-in
  medium: Fluid-Sludge (explicit connection)
  - selected OutletConnectionPoint clarifier-sludge
  - selected InletConnectionPoint sludge-pump-in
  - sludge-draw.medium = Fluid-Sludge

The explanation gives the chosen points, the medium, and the reason. The same values are on .medium, .medium_reason, and .evidence.

resolved.connection(handle) returns the concrete connection:

pipe = resolved.connection(sludge)
pipe, pipe.medium, list(pipe.connects_at)

(<Pipe 'sludge-draw'>,
 <enum Fluid-Sludge>,
 [<OutletConnectionPoint 'clarifier-sludge'>,
  <InletConnectionPoint 'sludge-pump-in'>])

Derived points and pipes exist only in the snapshot. Before resolution, equipment.ports holds points that already exist and declared ports. After, the snapshot’s copy holds every concrete point:

list(sludge_pump.ports)

[]

list(resolved["sludge-pump"].ports)

[<InletConnectionPoint 'sludge-pump-in'>,
 <OutletConnectionPoint 'sludge-pump-out'>]

The snapshot can be walked along its pipes. through= follows only pipes carrying one medium, here the sludge line:

start = resolved["clarifier"]
start.find_all_next(), start.find_all_next(through={"medium": values.Fluid_Sludge})

([<RapidSandFiltrationStage 'sand-filter'>,
  <Pump 'sludge-pump'>,
  <GravityThickener 'thickener'>],
 [<Pump 'sludge-pump'>, <GravityThickener 'thickener'>])

See Find objects in a model for the other search and walk methods.

Any edit to the model invalidates the snapshot.

Describe a connection#

Set properties on the handle. They are copied to the resolved connection:

sludge.label = "Clarifier sludge draw"
spent = connect(sand_filter.backwash_waste, recovery_tank, name="spent-backwash")
spent.has_role.add(values.Role_Backwash)

Roles do not affect which medium the solver picks.

Add more points#

Shape requirements are minimums. Request an extra point with ports="new":

connect(source, mixer, ports="new")

Pair two ports that carry the same flow through a device. Pairing makes their media equal and records s223:pairedConnectionPoint:

feed_in = exchanger.port("feed-in", direction="in")
feed_out = exchanger.port("feed-out", direction="out")
feed_in.pair(feed_out)

Next#