"""Plain-text tracing of Modbus transactions (verbose/debug mode). Formats decoded requests and responses in human-readable form, e.g.:: REQ txn=0x0001 unit=1 FC03 Read Holding Registers addr=0x2002 count=1 RSP txn=0x0001 unit=1 FC03 -> 1 reg: [0x1234] EXC txn=0x0001 unit=1 FC03 -> 0x02 Illegal Data Address (addr=0x0100) Register addresses are annotated with their WAGO 750-362 meaning (process image areas, read-back mirrors, configuration/constant registers) so traces are directly readable during FAT bench work. """ from __future__ import annotations from . import pdu as pdu_mod from .mbap import MbapHeader # --- function code names ------------------------------------------------------- _FC_NAMES = { 0x01: "Read Coils", 0x02: "Read Discrete Inputs", 0x03: "Read Holding Registers", 0x04: "Read Input Registers", 0x05: "Write Single Coil", 0x06: "Write Single Register", 0x0B: "Get Comm Event Count", 0x0F: "Write Multiple Coils", 0x10: "Write Multiple Registers", 0x16: "Mask Write Register", 0x17: "Read/Write Multiple Registers", } _EXCEPTION_NAMES = { 0x01: "Illegal Function", 0x02: "Illegal Data Address", 0x03: "Illegal Data Value", 0x04: "Slave Device Failure", 0x05: "Acknowledge", 0x06: "Server Busy", 0x08: "Memory Parity Error", 0x0A: "Gateway Path Unavailable", 0x0B: "Gateway Target Failed", } # --- WAGO register-map annotations (750-362 manual §11.2.4/§11.2.5) ------------- def describe_address(address: int, count: int = 1, *, write: bool = False) -> str: """Annotate a word address with its WAGO meaning, if any. `write` selects the mapping per the manual: Table 119 (reads) maps 0x0000-0x00FF to the INPUT area (%IW), while Table 120 (writes) maps the same range to the OUTPUT area (%QW); mirrors swap accordingly. """ end = address + count - 1 def area(name: str, note: str) -> str: if count == 1: return f"{name} {note}" return f"{name} {note} [{address:#06x}..{end:#06x}]" if write: if address <= 0x00FF: return area("output area 1", f"(%QW{address})") if address <= 0x01FF: return "illegal area 0x0100-0x01FF" if address <= 0x02FF: return area("output area 1 mirror", f"(%QW{address - 0x0200})") if address <= 0x0FFF: return "illegal area 0x0300-0x0FFF" if address <= 0x2FFF: return f"config regs {address:#06x}" if address < 0x2000 \ else f"config/constant regs {address:#06x}" if address <= 0x5FFF: return "illegal area 0x3000-0x5FFF" if address <= 0x62FB: return area("output area 2", f"(%QW{256 + address - 0x6000})") if address <= 0x6FFF: return "illegal area 0x62FC-0x6FFF" if address <= 0x72FB: return area("output area 2 mirror", f"(%QW{256 + address - 0x7000})") return "illegal area 0x72FC-0xFFFF" if address <= 0x00FF: return area("input area 1", f"(%IW{address})") if address <= 0x01FF: return "illegal area 0x0100-0x01FF" if address <= 0x02FF: return area("output area 1 read-back", f"(%QW{address - 0x0200})") if address <= 0x0FFF: return "illegal area 0x0300-0x0FFF" if address <= 0x2FFF: return f"config/constant regs {address:#06x}" if address <= 0x5FFF: return "illegal area 0x3000-0x5FFF" if address <= 0x62FB: return area("input area 2", f"(%IW{256 + address - 0x6000})") if address <= 0x6FFF: return "illegal area 0x62FC-0x6FFF" if address <= 0x72FB: return area("output area 2 read-back", f"(%QW{256 + address - 0x7000})") return "illegal area 0x72FC-0xFFFF" def describe_bit_address(address: int, count: int = 1) -> str: """Annotate a bit address (digital channel number semantics).""" end = address + count - 1 if address < 0x0200: base = f"D/I {address}..{end} area 1" if count > 1 else f"channel {address} area 1" elif address < 0x0400: ch = address - 0x0200 base = (f"DO {ch}..{end - 0x0200} area 1 read-back" if count > 1 else f"DO {ch} area 1 read-back") elif address >= 0x9000 and address <= 0x95F7: ch = 513 + address - 0x9000 base = (f"DO {ch}..{ch + count - 1} area 2 read-back" if count > 1 else f"DO {ch} area 2 read-back") elif address >= 0x8000 and address <= 0x85F7: ch = 513 + address - 0x8000 base = (f"DI {ch}..{ch + count - 1} area 2" if count > 1 else f"DI {ch} area 2") else: base = "illegal bit area" return base def _fc_name(fc: int) -> str: return _FC_NAMES.get(fc, f"Unknown ({fc:#04x})") def _exc_name(code: int) -> str: return _EXCEPTION_NAMES.get(code, f"Unknown ({code:#04x})") def _prefix(header: MbapHeader) -> str: return f"txn=0x{header.transaction_id:04X} unit={header.unit_id}" # --- request formatting --------------------------------------------------------- def format_request(header: MbapHeader, request: pdu_mod.Request) -> str: """One- or two-line plain-text description of a decoded request.""" fc = request.function_code head = f"REQ {_prefix(header)} FC{fc:02X} {_fc_name(fc)}" r = request if isinstance(r, pdu_mod.ReadRequest): is_bits = fc in (pdu_mod.FC_READ_COILS, pdu_mod.FC_READ_DISCRETE_INPUTS) where = (describe_bit_address(r.address, r.count) if is_bits else describe_address(r.address, r.count)) return f"{head}\n addr={r.address:#06x} count={r.count} [{where}]" if isinstance(r, pdu_mod.WriteSingleCoilRequest): return (f"{head}\n addr={r.address:#06x} " f"value={'ON' if r.value else 'OFF'} " f"[{describe_bit_address(r.address)}]") if isinstance(r, pdu_mod.WriteSingleRegisterRequest): return (f"{head}\n addr={r.address:#06x} value={r.value:#06x} " f"[{describe_address(r.address, write=True)}]") if isinstance(r, pdu_mod.WriteMultipleCoilsRequest): vals = "".join("1" if v else "0" for v in r.values) return (f"{head}\n addr={r.address:#06x} values={vals} " f"[{describe_bit_address(r.address, len(r.values))}]") if isinstance(r, pdu_mod.WriteMultipleRegistersRequest): vals = ", ".join(f"{v:#06x}" for v in r.values) return (f"{head}\n addr={r.address:#06x} values=[{vals}] " f"[{describe_address(r.address, len(r.values), write=True)}]") if isinstance(r, pdu_mod.MaskWriteRegisterRequest): return (f"{head}\n addr={r.address:#06x} and={r.and_mask:#06x} " f"or={r.or_mask:#06x} [{describe_address(r.address, write=True)}]") if isinstance(r, pdu_mod.ReadWriteMultipleRegistersRequest): return (f"{head}\n read addr={r.read_address:#06x} count={r.read_count} " f"[{describe_address(r.read_address, r.read_count)}] | " f"write addr={r.write_address:#06x} values={list(r.values)} " f"[{describe_address(r.write_address, write=True)}]") if isinstance(r, pdu_mod.GetCommEventCountRequest): return head return f"{head} (no formatter)" # pragma: no cover def format_response(header: MbapHeader, response: pdu_mod.Response) -> str: """Plain-text description of a response or exception.""" prefix = f"{_prefix(header)}" if isinstance(response, pdu_mod.ExceptionResponse): return (f"EXC {prefix} FC{response.function_code:02X} " f"-> code=0x{response.exception_code:02X} " f"{_exc_name(response.exception_code)}") resp = response if isinstance(resp, pdu_mod.ReadResponse): if resp.bits: return (f"RSP {prefix} FC{resp.function_code:02X} " f"-> {len(resp.bits)} bits: " f"{''.join('1' if b else '0' for b in resp.bits)}") return (f"RSP {prefix} FC{resp.function_code:02X} " f"-> {len(resp.registers)} regs: " f"[{', '.join(f'{v:#06x}' for v in resp.registers)}]") if isinstance(resp, pdu_mod.WriteSingleCoilResponse): return f"RSP {prefix} FC{resp.function_code:02X} -> echo" if isinstance(resp, pdu_mod.WriteSingleRegisterResponse): return f"RSP {prefix} FC{resp.function_code:02X} -> echo" if isinstance(resp, pdu_mod.WriteMultipleResponse): return (f"RSP {prefix} FC{resp.function_code:02X} " f"-> wrote {resp.count} items") if isinstance(resp, pdu_mod.MaskWriteRegisterResponse): return f"RSP {prefix} FC{resp.function_code:02X} -> echo" if isinstance(resp, pdu_mod.GetCommEventCountResponse): return (f"RSP {prefix} FC{resp.function_code:02X} " f"-> events={resp.event_count}") return f"RSP {prefix} -> (no formatter)" # pragma: no cover def format_raw(header: MbapHeader, pdu: bytes) -> str: """Raw-frame line for structurally invalid PDUs (cannot be decoded).""" hexed = pdu.hex(" ") if pdu else "(empty)" return f"RAW {prefix_stub(header)} PDU[{len(pdu)}]: {hexed}" def prefix_stub(header: MbapHeader) -> str: return _prefix(header)