69 lines
2.7 KiB
Python
69 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import time
|
|
|
|
from hil_common import configured_port, load_config, main_guard, serial_module
|
|
|
|
|
|
def frame(seconds: str = "00000064", include_energy: bool = True, terminator: bool = True) -> bytes:
|
|
lines = ["/DD3HIL", f"0-0:96.8.0*255({seconds})"]
|
|
if include_energy:
|
|
lines.append("1-0:1.8.0*255(001234.567*kWh)")
|
|
lines.extend([
|
|
"1-0:16.7.0*255(000950*W)",
|
|
"1-0:36.7.0*255(000500*W)",
|
|
"1-0:56.7.0*255(000450*W)",
|
|
"1-0:76.7.0*255(000000*W)",
|
|
])
|
|
text = "\r\n".join(lines) + "\r\n"
|
|
if terminator:
|
|
text += "!\r\n"
|
|
return text.encode("ascii")
|
|
|
|
|
|
def fixture(name: str) -> list[tuple[bytes, float]]:
|
|
valid = frame()
|
|
fixtures = {
|
|
"valid": [(valid, 0)],
|
|
"timeout": [(b"/DD3HIL\r\n1-0:1.8.0*255(1*kWh)", 2.0)],
|
|
"truncated": [(valid[:40], 2.0)],
|
|
"missing_terminator": [(frame(terminator=False), 2.0)],
|
|
"malformed_obis": [(valid.replace(b"1-0:1.8.0", b"1-0:X.Y.Z"), 0)],
|
|
"missing_required": [(frame(include_energy=False), 0)],
|
|
"invalid_meter_seconds": [(frame(seconds="GGGGGGGG"), 0)],
|
|
"timestamp_rollback": [(frame(seconds="00000100"), 0.2), (frame(seconds="00000080"), 0)],
|
|
"timestamp_jump": [(frame(seconds="00000100"), 0.2), (frame(seconds="00001000"), 0)],
|
|
"oversized": [(b"/" + b"A" * 600 + b"!\r\n", 0)],
|
|
"malformed_then_valid": [(valid.replace(b"1-0:1.8.0", b"1-0:X.Y.Z"), 0.2), (valid, 0)],
|
|
}
|
|
if name == "slow_inter_byte":
|
|
return [(bytes([byte]), 0.02) for byte in valid]
|
|
return fixtures[name]
|
|
|
|
|
|
def main() -> int:
|
|
choices = ["valid", "timeout", "truncated", "missing_terminator", "malformed_obis",
|
|
"missing_required", "invalid_meter_seconds", "timestamp_rollback", "timestamp_jump",
|
|
"oversized", "slow_inter_byte", "malformed_then_valid"]
|
|
parser = argparse.ArgumentParser(description="Drive an explicitly approved isolated 9600 7E1 meter simulator")
|
|
parser.add_argument("fixture", choices=choices)
|
|
args = parser.parse_args()
|
|
config = load_config()
|
|
port = configured_port(config, "meter_simulator", "access")
|
|
serial = serial_module()
|
|
with serial.Serial(str(port.device), 9600, bytesize=serial.SEVENBITS, parity=serial.PARITY_EVEN,
|
|
stopbits=serial.STOPBITS_ONE, timeout=1, write_timeout=2, exclusive=True) as handle:
|
|
for payload, pause in fixture(args.fixture):
|
|
handle.write(payload)
|
|
handle.flush()
|
|
if pause:
|
|
time.sleep(pause)
|
|
print(f"Sent {args.fixture} through approved isolated simulator {port.device}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_guard(main)
|