#!/usr/bin/env python3 """Example: MIDAS-frontend-style loop using the Python MSCB module. Features: - periodic integer-variable polling - optional periodic integer write - reconnect and resume on communication errors """ from __future__ import annotations import argparse import ctypes from datetime import datetime from pathlib import Path import sys import time from types import ModuleType class MscbError(RuntimeError): """Raised when an MSCB operation fails.""" def _load_mscb() -> ModuleType: python_dir = Path(__file__).resolve().parents[1] / "python" python_dir_str = str(python_dir) if python_dir_str not in sys.path: sys.path.insert(0, python_dir_str) import mscb return mscb def _ts() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _parse_int(value: str) -> int: return int(value, 0) def _parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("-d", "--device", default="mscb440", help="Ethernet submaster hostname or IP[:port]") parser.add_argument("-p", "--password", default="", help="Optional submaster password") parser.add_argument("-n", "--node", type=_parse_int, default=6, help="MSCB node address") parser.add_argument("--read-index", type=_parse_int, default=160, help="Integer variable index to poll") parser.add_argument("--read-period", type=float, default=1.0, help="Read period in seconds") parser.add_argument("--write-index", type=_parse_int, default=None, help="Optional integer variable to write") parser.add_argument("--write-value", type=_parse_int, default=None, help="Optional integer value to write") parser.add_argument("--write-period", type=float, default=10.0, help="Write period in seconds") parser.add_argument("--max-iterations", type=int, default=0, help="Stop after N reads (0=infinite)") parser.add_argument("--reconnect-delay", type=float, default=2.0, help="Delay before reconnecting") return parser.parse_args(argv) def _validate_args(args: argparse.Namespace) -> None: if (args.write_index is None) != (args.write_value is None): raise ValueError("--write-index and --write-value must be set together") if args.read_period <= 0: raise ValueError("--read-period must be > 0") if args.write_period <= 0: raise ValueError("--write-period must be > 0") if args.reconnect_delay < 0: raise ValueError("--reconnect-delay must be >= 0") if args.max_iterations < 0: raise ValueError("--max-iterations must be >= 0") def _check(mscb: ModuleType, status: int, operation: str) -> None: if status != mscb.MSCB_SUCCESS: raise MscbError(f"{operation} failed with MSCB status {status}") def _variable_info(mscb: ModuleType, fd: int, node: int, index: int): info = mscb.MSCB_INFO_VAR() status = mscb.mscb_info_variable(fd, node, index, ctypes.byref(info)) _check(mscb, status, f"metadata read for node {node}, variable {index}") if int(info.flags) & 1 or int(info.unit) in (55, 56): raise MscbError(f"variable {index} is not an integer") if not 1 <= int(info.width) <= 8: raise MscbError(f"variable {index} has unsupported integer width {info.width}") return info def _read_integer(mscb: ModuleType, fd: int, node: int, index: int) -> int: info = _variable_info(mscb, fd, node, index) size = ctypes.c_int(int(info.width)) data = ctypes.create_string_buffer(size.value) status = mscb.mscb_read(fd, node, index, data, ctypes.byref(size)) _check(mscb, status, f"read of node {node}, variable {index}") signed = bool(int(info.flags) & (1 << 1)) return int.from_bytes(data.raw[: size.value], "little", signed=signed) def _write_integer(mscb: ModuleType, fd: int, node: int, index: int, value: int) -> None: info = _variable_info(mscb, fd, node, index) width = int(info.width) signed = bool(int(info.flags) & (1 << 1)) try: payload = int(value).to_bytes(width, "little", signed=signed) except OverflowError as exc: raise MscbError(f"value {value} does not fit in variable {index}") from exc data = ctypes.create_string_buffer(payload, len(payload)) status = mscb.mscb_write(fd, node, index, data, len(payload)) _check(mscb, status, f"write of node {node}, variable {index}") def _connect(mscb: ModuleType, args: argparse.Namespace) -> int: fd = mscb.mscb_init(args.device, len(args.device), args.password, 0) if fd < 0: raise MscbError(f"mscb_init({args.device!r}) failed with status {fd}") status = mscb.mscb_ping(fd, args.node, 0, 1) if status != mscb.MSCB_SUCCESS: mscb.mscb_exit(fd) raise MscbError(f"node {args.node} does not respond (status {status})") return fd def main(argv: list[str]) -> int: args = _parse_args(argv) try: _validate_args(args) except ValueError as exc: print(f"Configuration error: {exc}", file=sys.stderr) return 2 mscb = _load_mscb() print(f"{_ts()} Starting frontend loop: device={args.device} node={args.node} read_index={args.read_index}") if args.write_index is not None: print( f"{_ts()} Periodic write enabled: index={args.write_index} " f"value={args.write_value} every {args.write_period:.3f}s" ) else: print(f"{_ts()} Periodic write disabled") fd = -1 read_count = 0 next_read_at = time.monotonic() next_write_at = time.monotonic() try: while not args.max_iterations or read_count < args.max_iterations: try: if fd < 0: fd = _connect(mscb, args) print(f"{_ts()} Connected to {args.device}; node {args.node} responds") next_read_at = time.monotonic() next_write_at = time.monotonic() now = time.monotonic() if now < next_read_at: time.sleep(min(0.05, next_read_at - now)) continue value = _read_integer(mscb, fd, args.node, args.read_index) read_count += 1 print(f"{_ts()} READ #{read_count:06d} idx={args.read_index} value={value}") next_read_at = now + args.read_period if args.write_index is not None and now >= next_write_at: _write_integer(mscb, fd, args.node, args.write_index, args.write_value) print(f"{_ts()} WRITE idx={args.write_index} value={args.write_value}") next_write_at = now + args.write_period except MscbError as exc: print(f"{_ts()} MSCB error: {exc}", file=sys.stderr) if fd >= 0: mscb.mscb_exit(fd) fd = -1 print(f"{_ts()} Reconnecting in {args.reconnect_delay:.2f}s") time.sleep(args.reconnect_delay) print(f"{_ts()} Reached max iterations ({args.max_iterations}), stopping") except KeyboardInterrupt: print(f"{_ts()} Stopped by user") finally: if fd >= 0: mscb.mscb_exit(fd) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))