Digital twin
A digital twin in Aether is usually a Sigil graph plus one or more Weaves:
- Writable Sigils — Commands, setpoints, or scenario knobs operators (or tools) adjust over OPC UA, REST, or MCP. They behave like inputs to your model or supervisory layer.
- Read-only Sigils (
writable=False) — Simulated state: temperatures, pressures, levels, or any outputs your model updates. Clients can subscribe and read them like real telemetry, but they are not meant to be overwritten from the edge. - Weave — An async callback at a fixed
cycle_timethat reads the current Sigil values, runs your equations or state machine, thenwrites the simulated Sigils. That is your plant model, observer, or soft sensor loop.
You host everything on the embedded OPC UA server by omitting per-Sigil opc_ua_url, so HMIs and agents see the same namespace as your HTTP and MCP clients. Set Nexus(opc_ua_url=...) to the endpoint clients use (for example opc.tcp://localhost:4840 on a workstation, or opc.tcp://0.0.0.0:4840 in Docker when publishing port 4840).
The same pattern applies to other domains (HVAC, line speed, tank farm, etc.): rename NodeIds, split or merge Sigils, and replace the body of the weave with your own dynamics.
Example: simplified steam boiler
The listing below illustrates the twin pattern with a compact steam drum toy model: three writable “operator” setpoints (fuel, feedwater, steam demand) and two read-only simulated variables (water temperature, steam pressure). The boiler_physics callback runs every 100 ms; swap this block for your process while keeping the Sigil + Weave structure.
from aether.nexus import Nexus
from aether.sigil import Sigil
from aether.weave import Weave
nexus = Nexus(
opc_ua_url="opc.tcp://10.128.0.32:4840"
)
# Control (writable setpoints)
fuel_setpoint = Sigil(
node_id="ns=2;s=Boiler.FuelSetpoint",
initial_value=45.0,
description="Fuel input command (0–100%). Scales heat added to the water side.",
)
feedwater = Sigil(
node_id="ns=2;s=Boiler.FeedWater",
initial_value=50.0,
description="Feedwater command (0–100%). Cooler makeup increases heat removal from the drum.",
)
steam_demand = Sigil(
node_id="ns=2;s=Boiler.SteamDemand",
initial_value=35.0,
description="Steam load request (0–100%). Drives enthalpy draw and turbine/header outflow in the model.",
)
# Process (simulated plant)
water_temp = Sigil(
node_id="ns=2;s=Boiler.WaterTemp",
initial_value=88.0,
description="Simulated bulk water temperature (°C). Written by the physics loop; not an operator setpoint.",
writable=False,
)
steam_pressure = Sigil(
node_id="ns=2;s=Boiler.SteamPressure",
initial_value=1.1,
description="Simulated steam drum pressure (bar gauge). Written by the physics loop; not an operator setpoint.",
writable=False,
)
async def boiler_physics():
T = float(await water_temp.read())
P = float(await steam_pressure.read())
fuel = float(await fuel_setpoint.read())
feed = float(await feedwater.read())
demand = float(await steam_demand.read())
fuel = max(0.0, min(100.0, fuel))
feed = max(0.0, min(100.0, feed))
demand = max(0.0, min(100.0, demand))
q_in = (fuel / 100.0) * 2200.0
q_out = (
40.0
+ (demand / 100.0) * 800.0 * (1.0 + 0.1 * P)
+ (feed / 100.0) * 30.0 * max(0.0, T - 30.0)
)
T += (q_in - q_out) / 4000.0
t_sat = 100.0 + 15.0 * max(0.0, P) ** 0.5
gen = max(0.0, (T - t_sat) * 0.02)
out = (demand / 100.0) * 0.9 * P**0.5 if P > 0.05 else 0.0
P += (gen - out) * 0.5
T = max(20.0, min(280.0, T))
P = max(0.05, min(12.0, P))
await water_temp.write(T)
await steam_pressure.write(P)
Weave(
label="boiler",
cycle_time=0.1,
callback=boiler_physics,
description="Boiler plant model: heat balance, saturation, steam generation, and pressure dynamics each cycle.",
)
if __name__ == "__main__":
nexus.start()
Run
Save the code to a file and run it:
python main.py
The HTTP API listens on 0.0.0.0:8000. Use GET /health for liveness. Interactive REST documentation is at /docs. MCP is mounted at /nexus.
Related
- Docker Compose to run the same pattern in a container.