jobe/wago-modbustcp-simulator-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 WIP

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-10-04 20:59:22 +02:00
parent f82e8d4be9
commit 68d3c42d79
12 changed files with 394 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wago_sim/modules.py
+42 -4
View File
@@ -196,6 +196,39 @@ class VirtualNode:
self.digital_in_bits = digital_in_bits
self.digital_out_bits = digital_out_bits
self._densify_ranges()
def _densify_ranges(self) -> None:
"""Complete the analog range map (FR-10.2.14, dense store).
Every analog channel gets an entry — defaults materialized —
and codes that no longer match an analog channel are pruned
(module list changed, FR-11.2). Digital codes never belong in
the range map (ranges apply to analog bars only).
"""
analog_codes = {self.io_code_at_slot("ai", loc.module_index, loc.channel - 1)
for loc in self._ai_channels} | \
{self.io_code_at_slot("ao", loc.module_index, loc.channel - 1)
for loc in self._ao_channels}
dense: dict[str, tuple[int, int]] = {}
for code in sorted(analog_codes):
spec = self.channel_ranges.get(code)
if spec is None or spec == self.DEFAULT_RANGE:
dense[code] = self.DEFAULT_RANGE
else:
dense[code] = spec
self.channel_ranges = dense
def set_channel_ranges(self, ranges: dict[str, tuple[int, int]]) -> None:
"""Replace all analog display ranges, then re-densify.
The store ends dense (FR-10.2.14): every analog channel of the
CURRENT layout present (missing entries become the default),
codes that no longer match an analog channel pruned.
"""
self.channel_ranges = dict(ranges)
self._densify_ranges()
def description(self) -> str:
"""Text for firmware-info register 0x2030 (module description)."""
return ", ".join(f"{i}:{m.article}" for i, m in enumerate(self.modules))
@@ -224,11 +257,16 @@ class VirtualNode:
def channel_range(self, kind: str, module_slot: int, channel: int) -> tuple[int, int]:
"""Display range (min, max) raw counts for an analog channel.
Defaults to (0, 4095) when unconfigured (FR-10.2.14).
Always defined for analog channels (dense store, FR-10.2.14):
the node materializes the default (0, 4095) at layout time,
so this returns the stored entry directly. Digital kinds are
rejected — ranges apply to analog bars only.
"""
return self.channel_ranges.get(
self.io_code_at_slot(kind, module_slot, channel), self.DEFAULT_RANGE
)
if kind not in ("ai", "ao"):
raise ValueError(f"channel_range() is analog-only, got kind {kind!r}")
return self.channel_ranges[
self.io_code_at_slot(kind, module_slot, channel)
]
def validate_channel_names(self) -> list[str]:
"""Names whose I/O code does not exist on this node (after a
src/wago_sim/scenario/plantenv.py
+45 -4
View File
@@ -25,10 +25,14 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from ..modules import VirtualNode
from .engine import ScenarioError
if TYPE_CHECKING:
from .engine import ScenarioEngine
@dataclass
class ChannelState:
@@ -40,6 +44,7 @@ class ChannelState:
io_code: str # e.g. "DI0.0"
name: str | None # configured application name, if any
value: object # bool for di/do, int/float for ai/ao
range: tuple[int, int] | None = None # (min, max) raw counts — analog only
@dataclass
@@ -62,6 +67,35 @@ class PlantState:
"""{io_code: value} for all channels of one kind."""
return {c.io_code: c.value for c in self.channels if c.kind == kind}
def channel(self, key: str) -> ChannelState:
"""Full ChannelState (name, range, ...) by I/O code or name.
Matches the I/O code first ("AI0.0"), then the configured
application name ("speed_setpoint", FR-10.2.11). KeyError
when neither matches — including a `None` name (unconfigured
channels match by code only).
"""
for ch in self.channels:
if ch.io_code == key:
return ch
for ch in self.channels:
if ch.name == key:
return ch
raise KeyError(key)
def channel_range(self, key: str) -> tuple[int, int]:
"""Display range (min, max) of an analog channel, by I/O code
or name (FR-10.2.14).
Always defined for AI/AO — the range store is dense — so
env models can unpack it without None-handling; digital
channels carry no range and raise KeyError.
"""
rng = self.channel(key).range
if rng is None:
raise KeyError(key)
return rng
class PlantEnv:
"""gym-style plant environment (FR-8.2.14-17).
@@ -75,9 +109,14 @@ class PlantEnv:
TICK_MS = 10.0
def __init__(self, node: VirtualNode | None = None, *,
engine=None) -> None:
engine: ScenarioEngine | None = None) -> None:
if node is None:
if engine is None:
raise ScenarioError("PlantEnv needs a node or an engine")
self.node: VirtualNode = engine.node
else:
self.node = node
self.engine = engine
self.node: VirtualNode = node if node is not None else engine.node
self.datastore = self.node.datastore
self.sim_time_ms = 0.0
# attachments (optional; the scenario layer wires these)
@@ -114,6 +153,8 @@ class PlantEnv:
io_code=code,
name=node.channel_names.get(code),
value=value,
range=node.channel_ranges[code]
if kind in ("ai", "ao") else None,
))
aliases = ({name: self._alias_value(name)
for name in (self._aliases or {})}
@@ -198,14 +239,14 @@ class PlantEnv:
if kind == "di":
ds.set_digital_input_channel(loc.bit + 1, bool(value))
elif kind == "ai":
raw = int(float(value) * scale + offset) & 0xFFFF
raw = int(float(value) * scale + offset) & 0xFFFF # type: ignore[arg-type]
ds.set_input_word(loc.word, raw)
elif kind == "do":
result = ds.write_bits(loc.bit, [bool(value)])
if isinstance(result, Exception):
raise ScenarioError(f"force() write to {key!r}: {result}")
else: # ao
raw = int(float(value) * scale + offset) & 0xFFFF
raw = int(float(value) * scale + offset) & 0xFFFF # type: ignore[arg-type]
result = ds.write_words(loc.word, [raw])
if isinstance(result, Exception):
raise ScenarioError(f"force() write to {key!r}: {result}")
src/wago_sim/webui/api.py
+25 -14
View File
@@ -14,7 +14,8 @@ from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
from ..app import SimulatorApp, rebuild_node
from ..modules import CATALOG, ChannelMismatch, ModuleUnknown, resolve_module
from ..modules import (CATALOG, ChannelMismatch, ModuleUnknown, VirtualNode,
resolve_module)
from ..scenario.engine import ScenarioError
from ..watchdog import RegisterWriteRefused
from . import state
@@ -417,7 +418,7 @@ async def put_module_config(
save_node_config(sim.node_config_path, modules,
name=Path(sim.node_config_path).stem,
channel_names=sim.node.channel_names,
channel_ranges=sim.node.channel_ranges)
channel_ranges=_nondefault_ranges(sim))
except OSError as exc:
raise HTTPException(
status_code=500,
@@ -485,6 +486,7 @@ async def put_channel_names(
sim.node_config_path, sim.node.modules,
name=Path(sim.node_config_path).stem,
channel_names=names,
channel_ranges=sim.node.channel_ranges,
)
except OSError as exc:
raise HTTPException(
@@ -520,8 +522,10 @@ async def put_channel_ranges(
"""Set per-channel analog display ranges (FR-10.2.14).
Payload: {"channel_ranges": {"AI4.0": {"min": 0, "max": 32000}, ...}}.
Codes are regex-validated; 0 <= min < max <= 65535 required; an
entry with min=0/max=4095 (the default) is dropped from the store.
Codes are regex-validated; -65534 <= min < max <= 65535 required.
The runtime store is DENSE (every analog channel present, default
0-4095 materialized); default-equal entries are kept in the store
but omitted from the persisted YAML (lean config).
"""
_require_write(request, sessions)
raw = payload.get("channel_ranges")
@@ -550,10 +554,9 @@ async def put_channel_ranges(
status_code=400,
detail=f"range for {code_up!r}: need -65534 <= min < max <= 65535",
)
if (lo, hi) == (0, 4095):
continue # default — keep the store clean
ranges[code_up] = (lo, hi)
sim.node.channel_ranges = ranges
ranges[code_up] = (lo, hi) # dense store keeps defaults
sim.node.set_channel_ranges(ranges)
persisted = _nondefault_ranges(sim)
if payload.get("persist", True) and sim.node_config_path:
from .persist import save_node_config
try:
@@ -561,7 +564,7 @@ async def put_channel_ranges(
sim.node_config_path, sim.node.modules,
name=Path(sim.node_config_path).stem,
channel_names=sim.node.channel_names,
channel_ranges=ranges,
channel_ranges=persisted,
)
except OSError as exc:
raise HTTPException(
@@ -570,7 +573,8 @@ async def put_channel_ranges(
) from None
return {
"ok": True,
"channel_ranges": {c: {"min": lo, "max": hi} for c, (lo, hi) in ranges.items()},
"channel_ranges": {c: {"min": lo, "max": hi}
for c, (lo, hi) in persisted.items()},
}
@@ -593,6 +597,14 @@ def _coupler_snapshot(sim: SimulatorApp) -> dict:
}
def _nondefault_ranges(sim: SimulatorApp) -> dict[str, tuple[int, int]]:
"""Dense store minus default entries — for YAML persistence and
the config export (lean config: only explicit overrides).
"""
return {code: rng for code, rng in sim.node.channel_ranges.items()
if rng != VirtualNode.DEFAULT_RANGE}
@router.get("/api/config/export")
async def export_config(
sim: SimulatorApp = Depends(get_sim),
@@ -606,7 +618,7 @@ async def export_config(
"modules": [m.article for m in sim.node.modules],
"channel_names": dict(sim.node.channel_names),
"channel_ranges": {c: {"min": lo, "max": hi}
for c, (lo, hi) in sim.node.channel_ranges.items()},
for c, (lo, hi) in _nondefault_ranges(sim).items()},
"coupler_settings": _coupler_snapshot(sim),
}
@@ -663,8 +675,7 @@ async def import_config(
if not -65534 <= lo < hi <= 65535:
raise HTTPException(status_code=400,
detail=f"range for {code_up!r}: need -65534 <= min < max <= 65535")
if (lo, hi) != (0, 4095):
ranges[code_up] = (lo, hi)
ranges[code_up] = (lo, hi) # defaults kept; the store is dense
coupler = payload.get("coupler_settings") or {}
# -- everything validated: apply atomically -------------------------
@@ -703,7 +714,7 @@ async def import_config(
sim.node_config_path, modules,
name=Path(sim.node_config_path).stem,
channel_names=names,
channel_ranges=ranges,
channel_ranges=_nondefault_ranges(sim),
)
except OSError as exc:
raise HTTPException(
src/wago_sim/webui/persist.py
+12 -5
View File
@@ -10,7 +10,7 @@ from __future__ import annotations
from pathlib import Path
from ..modules import ModuleType
from ..modules import ModuleType, VirtualNode
_HEADER = """\
# Virtual WAGO node configuration — saved from the web UI (SAVE CONFIG)
@@ -52,8 +52,15 @@ def save_node_config(
for code in sorted(channel_names):
lines.append(f" {code}: {channel_names[code]}\n")
if channel_ranges:
lines.append(_RANGES_HEADER)
for code in sorted(channel_ranges):
lo, hi = channel_ranges[code]
lines.append(f" {code}: {{ min: {lo}, max: {hi} }}\n")
# lean YAML: the runtime store is dense (every analog channel,
# FR-10.2.14) — persist only explicit, non-default entries so a
# hand-edited config stays minimal and defaults survive a
# future DEFAULT_RANGE change
explicit = {code: rng for code, rng in channel_ranges.items()
if rng != VirtualNode.DEFAULT_RANGE}
if explicit:
lines.append(_RANGES_HEADER)
for code in sorted(explicit):
lo, hi = explicit[code]
lines.append(f" {code}: {{ min: {lo}, max: {hi} }}\n")
Path(path).write_text("".join(lines), encoding="utf-8", newline="\n")
src/wago_sim/webui/state.py
+4 -3
View File
@@ -22,9 +22,10 @@ def node_layout(node: VirtualNode) -> NodeLayoutJSON:
for ch in mod.channels:
channel_no = int(ch.name.split()[-1]) - 1
io_code = f"{ch.kind.upper()}{per_type[ch.kind]}.{channel_no}"
lo, hi = (0, 4095)
if ch.kind in ("ai", "ao"):
lo, hi = node.channel_ranges.get(io_code, (0, 4095))
# dense range store (FR-10.2.14): every analog channel has
# an entry (defaults materialized at layout time)
lo, hi = node.channel_ranges[io_code] \
if ch.kind in ("ai", "ao") else (0, 4095)
channels.append(
{
"name": ch.name,