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

🚧 Change module numbering

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-10-04 13:05:47 +02:00
parent 7948b794f5
commit e5df0b04ed
17 changed files with 644 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_scenario_engine.py
+39 -27
View File
@@ -88,13 +88,15 @@ class TestAliases:
class TestModuleAddressing:
"""Module type + module number + channel number addressing (DI0.1).
The pump node fixture modules: [DI_8, DI_16, DO_8, DO_16, AI_2, AO_2]
(indices 0..5). Module addressing uses 0-based module positions and
0-based channel numbers on the module.
The pump node fixture modules: [DI_8, DI_16, DO_8, DO_16, AI_2, AO_2].
Module addressing (FR-4.3.11 v1.5) numbers modules 0-based WITHIN
THEIR TYPE: DI0=DI_8, DI1=DI_16, DO0=DO_8, DO1=DO_16, AI0=AI_2,
AO0=AO_2 — regardless of physical slot interleaving. Channels are
0-based on the module.
"""
def test_module_addressing_di(self, node, datastore):
# DI0.1: module slot 0 (DI_8), channel 1 (0-based) = 2nd DI
# DI0.1: DI module 0 (DI_8), channel 1 (0-based) = 2nd DI
scenario = _scenario()
scenario["inputs"]["second_di"] = {"type": "di", "module": 0, "channel": 1}
engine = ScenarioEngine(node, scenario)
@@ -104,35 +106,42 @@ class TestModuleAddressing:
assert engine.alias_value("second_di") is True
def test_module_addressing_across_modules(self, node, datastore):
# DI1.0: module slot 1 (DI_16), channel 0 = 9th DI globally
# (8 channels on module 0 come first)
# DI1.0: DI module 1 (DI_16, physical slot 1), channel 0 = 9th DI
scenario = _scenario()
scenario["inputs"]["di16_first"] = {"type": "di", "module": 1, "channel": 0}
engine = ScenarioEngine(node, scenario)
engine.set_alias("di16_first", True)
assert datastore.read_bits(8, 1) == [True]
def test_module_addressing_do(self, node, datastore):
# DO2.2: module slot 2 (DO_8), channel 2 = 3rd DO globally
def test_module_addressing_do_interleaved(self, node, datastore):
# DO0.2: DO module 0 is the DO_8 at PHYSICAL SLOT 2 — per-type
# numbering means module 0, not slot 2
scenario = _scenario()
scenario["outputs"]["third_do"] = {"type": "do", "module": 2, "channel": 2}
scenario["outputs"]["third_do"] = {"type": "do", "module": 0, "channel": 2}
engine = ScenarioEngine(node, scenario)
bit = engine.alias_location("third_do").bit
assert bit == 2
assert datastore.write_bits(bit, [True]) is None
assert engine.alias_value("third_do") is True
def test_module_addressing_ai_ao(self, node, datastore):
# AI4.1: module slot 4 (AI_2), channel 1 = 2nd AI word
def test_do_module_1_is_do16(self, node):
# DO1.0: DO module 1 is the DO_16 at physical slot 3
scenario = _scenario()
scenario["inputs"]["ai2"] = {"type": "ai", "module": 4, "channel": 1, "scale": 1.0}
scenario["outputs"]["x"] = {"type": "do", "module": 1, "channel": 0}
engine = ScenarioEngine(node, scenario)
assert engine.alias_location("x").bit == 8 # after DO_8's 8 channels
def test_module_addressing_ai_ao(self, node, datastore):
# AI0.1: AI module 0 (AI_2 at physical slot 4), channel 1 = 2nd AI word
scenario = _scenario()
scenario["inputs"]["ai2"] = {"type": "ai", "module": 0, "channel": 1, "scale": 1.0}
engine = ScenarioEngine(node, scenario)
engine.set_alias("ai2", 0x1234)
loc = node.channel("ai", 2) # global 2nd AI
assert datastore.get_input_word(loc.word) == 0x1234
# AO5.1: module slot 5 (AO_2), channel 1 = 2nd AO word
scenario["outputs"] = {"ao2": {"type": "ao", "module": 5, "channel": 1}}
# AO0.1: AO module 0 (AO_2 at physical slot 5), channel 1 = 2nd AO word
scenario["outputs"] = {"ao2": {"type": "ao", "module": 0, "channel": 1}}
engine2 = ScenarioEngine(node, scenario)
bit_word = node.channel("ao", 2).word
assert datastore.write_words(bit_word, [0xBEEF]) is None
@@ -149,31 +158,34 @@ class TestModuleAddressing:
def test_module_out_of_range_rejected(self, node):
scenario = _scenario()
scenario["inputs"]["bad"] = {"type": "di", "module": 99, "channel": 0}
with pytest.raises(IndexError, match="module 99"):
# the node has only 2 DI modules
scenario["inputs"]["bad"] = {"type": "di", "module": 2, "channel": 0}
with pytest.raises(IndexError, match="DI module 2"):
ScenarioEngine(node, scenario)
def test_channel_out_of_range_rejected(self, node):
# module 0 is a DI_8 -> channels 0-7 only
# DI module 0 is a DI_8 -> channels 0-7 only
scenario = _scenario()
scenario["inputs"]["bad"] = {"type": "di", "module": 0, "channel": 8}
with pytest.raises(IndexError, match="channel 8"):
ScenarioEngine(node, scenario)
def test_kind_mismatch_rejected(self, node):
# module 0 is DI-only -> DO addressing must fail
def test_module_kind_routes_to_own_type(self, node):
"""Per-type numbering (v1.5): 'do' module numbers address the DO
modules regardless of physical slot — DO0 is the DO_8 at slot 2.
A 'do' alias on module 1 (the DO_16) is valid and distinct."""
scenario = _scenario()
scenario["outputs"]["bad"] = {"type": "do", "module": 0, "channel": 0}
from wago_sim.modules import ChannelMismatch
with pytest.raises(ChannelMismatch):
ScenarioEngine(node, scenario)
scenario["outputs"]["do8"] = {"type": "do", "module": 0, "channel": 0}
scenario["outputs"]["do16"] = {"type": "do", "module": 1, "channel": 0}
engine = ScenarioEngine(node, scenario)
assert engine.alias_location("do8").bit == 0
assert engine.alias_location("do16").bit == 8
def test_analog_kind_on_digital_module_rejected(self, node):
# module 0 is DI-only -> AI addressing must fail
# per-type routing: 'ai' module 9 is out of range (only 1 AI module)
scenario = _scenario()
scenario["inputs"]["bad"] = {"type": "ai", "module": 0, "channel": 0}
from wago_sim.modules import ChannelMismatch
with pytest.raises(ChannelMismatch):
scenario["inputs"]["bad"] = {"type": "ai", "module": 9, "channel": 0}
with pytest.raises(IndexError):
ScenarioEngine(node, scenario)
tests/test_webui.py
+42 -38
View File
@@ -106,7 +106,7 @@ class TestManipulation:
async def test_toggle_do_forces_output_image(self, client, datastore):
# DO_8 sits at module slot 2 in the pump fixture
r = await client.post("/api/channels/do/2/0", json={"action": "toggle"})
r = await client.post("/api/channels/do/0/0", json={"action": "toggle"})
assert r.status_code == 200
# DO channel 1 is forced ON in the OUTPUT image (master-side)
assert datastore.digital_output_channel(1) is True
@@ -117,13 +117,13 @@ class TestManipulation:
async def test_ai_value_set(self, client, datastore):
# AI_2 sits at module slot 4 in the pump fixture (DI_8, DI_16,
# DO_8, DO_16, AI_2, AO_2); its first word is input word 0
r = await client.post("/api/channels/ai/4/0", json={"value": 200})
r = await client.post("/api/channels/ai/0/0", json={"value": 200})
assert r.status_code == 200
assert r.json()["value"] == 200
assert datastore.get_input_word(0) == 200
async def test_ao_force_writes_output_image(self, client, datastore):
r = await client.post("/api/channels/ao/5/0", json={"value": 1234})
r = await client.post("/api/channels/ao/0/0", json={"value": 1234})
assert r.status_code == 200
# written at the output image word 0 -> read back at +0x0200
assert datastore.read_words(0x0200, 1) == [1234]
@@ -133,9 +133,13 @@ class TestManipulation:
assert r.status_code == 404
async def test_kind_mismatch_400(self, client):
# DI_16 module (index 1) has no DO channels
r = await client.post("/api/channels/do/1/0", json={"action": "toggle"})
assert r.status_code == 400
# per-type numbering (v1.5): DO module 0 is the DO_8 — asking
# for a DO channel beyond its 8 channels is out of range (404);
# DO module 2 does not exist (only DO0, DO1)
r = await client.post("/api/channels/do/2/0", json={"action": "toggle"})
assert r.status_code == 404
r = await client.post("/api/channels/do/0/9", json={"action": "toggle"})
assert r.status_code in (400, 404) # channel out of module range
class TestDiagnostics:
@@ -319,7 +323,7 @@ class TestModuleConfig:
class TestChannelNames:
async def test_names_in_layout_and_values(self, client, app):
r = await client.put("/api/config/channel-names", json={
"channel_names": {"DI0.0": "start_button", "AI4.0": "tank_level"},
"channel_names": {"DI0.0": "start_button", "AI0.0": "tank_level"},
"persist": False,
})
assert r.status_code == 200
@@ -331,23 +335,23 @@ class TestChannelNames:
# values snapshot carries it too
r = await client.get("/api/values")
entry = next(v for v in r.json()["values"]
if (v["kind"], v["module"], v["channel"]) == ("ai", 4, 0))
if (v["kind"], v["module"], v["channel"]) == ("ai", 0, 0))
assert entry["name"] == "tank_level"
async def test_names_persist_to_yaml(self, client, app, tmp_path):
cfg_file = tmp_path / "node.yaml"
app.node_config_path = str(cfg_file)
r = await client.put("/api/config/channel-names", json={
"channel_names": {"DO2.0": "pump_relay"}, "persist": True,
"channel_names": {"DO0.0": "pump_relay"}, "persist": True,
})
assert r.status_code == 200
text = cfg_file.read_text(encoding="utf-8")
assert "channel_names:" in text
assert "DO2.0: pump_relay" in text
assert "DO0.0: pump_relay" in text
# and it loads back
from wago_sim.scenario.dsl import load_yaml, load_channel_names
names = load_channel_names(load_yaml(cfg_file))
assert names == {"DO2.0": "pump_relay"}
assert names == {"DO0.0": "pump_relay"}
async def test_invalid_code_rejected(self, client):
r = await client.put("/api/config/channel-names", json={
@@ -464,21 +468,21 @@ class TestChannelRanges:
async def test_put_ranges_and_layout(self, client, app):
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0, "max": 32000},
"AO5.0": {"min": 400, "max": 2000}},
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000},
"AO0.0": {"min": 400, "max": 2000}},
"persist": False,
})
assert r.status_code == 200
# layout reflects the configured range
r = await client.get("/api/node")
ai = next(c for m in r.json()["modules"] if m["index"] == 4
for c in m["channels"] if c["kind"] == "ai" and c["address"] == "AI4.0")
for c in m["channels"] if c["kind"] == "ai" and c["address"] == "AI0.0")
assert ai["range"] == {"min": 0, "max": 32000}
assert app.node.channel_range("ao", 5, 0) == (400, 2000)
assert app.node.channel_ranges["AO0.0"] == (400, 2000)
async def test_default_range_dropped_from_store(self, client, app):
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0, "max": 4095}}, # == default
"channel_ranges": {"AI0.0": {"min": 0, "max": 4095}}, # == default
"persist": False,
})
assert r.status_code == 200
@@ -490,71 +494,71 @@ class TestChannelRanges:
"channel_ranges": {"XX4.0": {"min": 0, "max": 100}}, **base})
assert r.status_code == 400 # bad code
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 100, "max": 50}}, **base})
"channel_ranges": {"AI0.0": {"min": 100, "max": 50}}, **base})
assert r.status_code == 400 # min >= max
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0, "max": 70000}}, **base})
"channel_ranges": {"AI0.0": {"min": 0, "max": 70000}}, **base})
assert r.status_code == 400 # > 65535
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0}}, **base})
"channel_ranges": {"AI0.0": {"min": 0}}, **base})
assert r.status_code == 400 # missing max
async def test_negative_range_accepted(self, client, app):
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": -32000, "max": 32000}},
"channel_ranges": {"AI0.0": {"min": -32000, "max": 32000}},
"persist": False,
})
assert r.status_code == 200
assert app.node.channel_range("ai", 4, 0) == (-32000, 32000)
assert app.node.channel_ranges["AI0.0"] == (-32000, 32000)
# layout exposes the signed range
r = await client.get("/api/node")
ai = next(c for m in r.json()["modules"] if m["index"] == 4
for c in m["channels"] if c["address"] == "AI4.0")
for c in m["channels"] if c["address"] == "AI0.0")
assert ai["range"] == {"min": -32000, "max": 32000}
async def test_range_floor_is_minus_65534(self, client):
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": -65535, "max": 0}}, "persist": False})
"channel_ranges": {"AI0.0": {"min": -65535, "max": 0}}, "persist": False})
assert r.status_code == 400 # below -65534
async def test_signed_value_display_and_write(self, client, app, datastore):
# channel with a negative-min range -> signed two's-complement view
await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": -32000, "max": 32000}},
"channel_ranges": {"AI0.0": {"min": -32000, "max": 32000}},
"persist": False})
r = await client.post("/api/channels/ai/4/0", json={"value": -2000})
r = await client.post("/api/channels/ai/0/0", json={"value": -2000})
assert r.status_code == 200
# raw image holds the two's-complement word
assert datastore.get_input_word(0) == (-2000) & 0xFFFF
# values API shows the signed number again
r = await client.get("/api/values")
entry = next(v for v in r.json()["values"]
if (v["kind"], v["module"], v["channel"]) == ("ai", 4, 0))
if (v["kind"], v["module"], v["channel"]) == ("ai", 0, 0))
assert entry["value"] == -2000
async def test_negative_ao_force(self, client, datastore):
r = await client.post("/api/channels/ao/5/0", json={"value": -3000})
r = await client.post("/api/channels/ao/0/0", json={"value": -3000})
assert r.status_code == 200
assert datastore.get_output_word(0) == (-3000) & 0xFFFF
# unsigned AO (default range) still displays raw
r = await client.get("/api/values")
entry = next(v for v in r.json()["values"]
if (v["kind"], v["module"], v["channel"]) == ("ao", 5, 0))
if (v["kind"], v["module"], v["channel"]) == ("ao", 0, 0))
assert entry["value"] == (-3000) & 0xFFFF
async def test_value_bounds_rejected(self, client):
r = await client.post("/api/channels/ai/4/0", json={"value": -65535})
r = await client.post("/api/channels/ai/0/0", json={"value": -65535})
assert r.status_code == 400
r = await client.post("/api/channels/ai/4/0", json={"value": 65536})
r = await client.post("/api/channels/ai/0/0", json={"value": 65536})
assert r.status_code == 400
r = await client.post("/api/channels/ai/4/0", json={"value": "banana"})
r = await client.post("/api/channels/ai/0/0", json={"value": "banana"})
assert r.status_code == 400
async def test_ranges_persist_to_yaml(self, client, app, tmp_path):
cfg_file = tmp_path / "node.yaml"
app.node_config_path = str(cfg_file)
r = await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0, "max": 32000}}, "persist": True,
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000}}, "persist": True,
})
assert r.status_code == 200
text = cfg_file.read_text(encoding="utf-8")
@@ -563,18 +567,18 @@ class TestChannelRanges:
# loads back
from wago_sim.scenario.dsl import load_yaml, load_channel_ranges
ranges = load_channel_ranges(load_yaml(cfg_file))
assert ranges == {"AI4.0": (0, 32000)}
assert ranges == {"AI0.0": (0, 32000)}
async def test_ranges_survive_module_save(self, client, app, tmp_path):
app.node_config_path = str(tmp_path / "node.yaml")
await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 0, "max": 32000}}, "persist": True,
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000}}, "persist": True,
})
r = await client.put("/api/config/modules", json={
"modules": ["750-430", "750-467"], "persist": True,
})
assert r.status_code == 200
# AI_2 is at slot 1 now; the AI4.0 range became stale -> carried
# AI_2 is at slot 1 now; the AI0.0 range became stale -> carried
# over but no longer matches a channel; the store keeps it (names
# semantics differ: stale name codes are dropped on import only)
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
@@ -582,14 +586,14 @@ class TestChannelRanges:
async def test_export_import_carries_ranges(self, client, app, tmp_path):
await client.put("/api/config/channel-ranges", json={
"channel_ranges": {"AI4.0": {"min": 600, "max": 2000}}, "persist": False,
"channel_ranges": {"AI0.0": {"min": 600, "max": 2000}}, "persist": False,
})
exported = (await client.get("/api/config/export")).json()
assert exported["channel_ranges"]["AI4.0"] == {"min": 600, "max": 2000}
assert exported["channel_ranges"]["AI0.0"] == {"min": 600, "max": 2000}
app.node_config_path = str(tmp_path / "node.yaml")
r = await client.post("/api/config/import", json=exported)
assert r.status_code == 200
assert app.node.channel_range("ai", 4, 0) == (600, 2000)
assert app.node.channel_ranges["AI0.0"] == (600, 2000)
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
assert "min: 600, max: 2000" in text