"""WAGO 750-362 data model: process images and address-map dispatch. Implements the register mapping tables of the 750-362 manual §11.2.4: Word area: 0x0000-0x00FF input image area 1 (first 256 words) [R] 0x0100-0x01FF illegal [R/W] 0x0200-0x02FF output image area 1 readback (+0x0200) [R, mirrored W] 0x0300-0x0FFF illegal [R/W] 0x1000-0x2FFF configuration registers [R/W] 0x3000-0x5FFF illegal [R/W] 0x6000-0x62FB input image area 2 (words 256-1020) [R] 0x62FC-0x6FFF illegal [R/W] 0x7000-0x72FB output image area 2 readback (+0x1000) [R, mirrored W] 0x72FC-0xFFFF illegal [R/W] Word writes (FC6/FC16) land in the output image: 0x0000-0x00FF output image area 1 0x0200-0x02FF output image area 1 (mirror accepted) 0x6000-0x62FB output image area 2 0x7000-0x72FB output image area 2 (mirror accepted) Bit area (FC1/2/5/15), addresses are *digital channel numbers*: 0x0000-0x01FF digital inputs 1-512 [R] 0x0200-0x03FF digital outputs 1-512 [R/W] 0x0800-0x85F7 digital inputs 513-2039 [R] 0x9000-0x95F7 digital outputs 513-2039 [R/W] everything else illegal """ from __future__ import annotations from collections.abc import Callable from typing import Protocol from .pdu import ( EXC_ILLEGAL_DATA_ADDRESS, EXC_ILLEGAL_DATA_VALUE, ExceptionResponse, MaskWriteRegisterRequest, ReadRequest, ReadWriteMultipleRegistersRequest, WriteMultipleCoilsRequest, WriteMultipleRegistersRequest, WriteSingleCoilRequest, WriteSingleRegisterRequest, ) # --- Word area geometry ----------------------------------------------------- AREA1_INPUT_WORDS = 256 # 0x0000-0x00FF AREA2_INPUT_WORDS = 764 # 0x6000-0x62FB (%IW256-%IW1020) AREA1_OUTPUT_WORDS = 256 # 0x0200-0x02FF mirror of %QW0-255 AREA2_OUTPUT_WORDS = 764 # 0x7000-0x72FB mirror of %QW256-1020 WORD_IN1_BASE = 0x0000 WORD_IN2_BASE = 0x6000 WORD_OUT1_BASE = 0x0000 WORD_OUT1_MIRROR_BASE = 0x0200 WORD_OUT2_BASE = 0x6000 WORD_OUT2_MIRROR_BASE = 0x7000 WORD_CONFIG_BASE = 0x1000 WORD_CONFIG_END = 0x2FFF # --- Bit area geometry ------------------------------------------------------ BIT_IN1_BASE = 0x0000 # 0x0000-0x01FF DI 1-512 BIT_IN1_COUNT = 512 BIT_OUT1_BASE = 0x0200 # 0x0200-0x03FF DO 1-512 BIT_OUT1_COUNT = 512 BIT_IN2_BASE = 0x8000 # 0x8000-0x85F7 DI 513-2039 BIT_IN2_COUNT = 1527 BIT_OUT2_BASE = 0x9000 # 0x9000-0x95F7 DO 513-2039 BIT_OUT2_COUNT = 1527 DIAG_READ_LIMIT = 125 DIAG_WRITE_LIMIT = 100 COIL_READ_LIMIT = 2000 COIL_WRITE_LIMIT = 800 class AddressError(Exception): """Modbus address outside any mapped area (exception 0x02).""" def __init__(self, address: int) -> None: self.address = address super().__init__(f"illegal address: {address:#06x}") class ConfigRegisterError(Exception): """A configuration-register operation was rejected.""" class ConfigRegisters(Protocol): """Interface to the configuration-register bank (0x1000-0x2FFF).""" def read_config(self, address: int, count: int) -> list[int]: ... def write_config(self, address: int, values: list[int]) -> None: ... class WagoDatastore: """Holds both process images and dispatches word/bit accesses. The process images are plain word arrays; digital data is packed into words using the Intel format (manual §7.3.3): channel 1 = bit 0 of the low byte D0. """ def __init__(self, *, config_bank: ConfigRegisters) -> None: self._config = config_bank self._in_words = [0] * (AREA1_INPUT_WORDS + AREA2_INPUT_WORDS) self._out_words = [0] * (AREA1_OUTPUT_WORDS + AREA2_OUTPUT_WORDS) self._listeners: list[Callable[[], None]] = [] # Digital channels are byte-packed AFTER the complex (analog) # modules in the image (manual §7.3.2.1): bit address g (channel # number - 1) lives at word offset + g//16, bit g%16. self._digital_in_word_offset = 0 self._digital_out_word_offset = 0 def set_digital_word_offsets(self, in_words: int, out_words: int) -> None: """Tell the datastore where the byte-packed digital area starts. Called by the VirtualNode after layout: `in_words`/`out_words` are the total analog (word-oriented) module word counts. """ self._digital_in_word_offset = in_words self._digital_out_word_offset = out_words # -- subscription (scenario engine observes output writes) -------------- def add_listener(self, fn: Callable[[], None]) -> None: self._listeners.append(fn) def _notify(self) -> None: for fn in self._listeners: fn() # -- process image access (scenario engine / node side) ------------------ @property def input_words(self) -> list[int]: """Simulated inputs (PII), indices 0..1019 (word 0..255 area 1).""" return self._in_words @property def output_words(self) -> list[int]: """PLC-written outputs (PIO), indices 0..1019.""" return self._out_words def set_input_word(self, index: int, value: int) -> None: self._validate_index(index, len(self._in_words)) self._in_words[index] = value & 0xFFFF def get_input_word(self, index: int) -> int: self._validate_index(index, len(self._in_words)) return self._in_words[index] def set_input_bit(self, index: int, bit: int, value: bool) -> None: self._validate_index(index, len(self._in_words)) if value: self._in_words[index] |= 1 << bit else: self._in_words[index] &= ~(1 << bit) & 0xFFFF def get_input_bit(self, index: int, bit: int) -> bool: return bool(self._in_words[index] & (1 << bit)) def get_output_bit(self, index: int, bit: int) -> bool: return bool(self._out_words[index] & (1 << bit)) def get_output_word(self, index: int) -> int: return self._out_words[index] def clear_outputs(self) -> None: """Watchdog timeout reaction: analog->0, digital->OFF.""" self._out_words = [0] * len(self._out_words) self._notify() def reset_images(self) -> None: """Reconfiguration (§11): zero both images before a layout rebuild. The arrays keep their full 750-362 geometry (256+764 words per direction); the VirtualNode re-applies the digital word offsets and the module layout on top. """ self._in_words = [0] * (AREA1_INPUT_WORDS + AREA2_INPUT_WORDS) self._out_words = [0] * (AREA1_OUTPUT_WORDS + AREA2_OUTPUT_WORDS) self._digital_in_word_offset = 0 self._digital_out_word_offset = 0 self._notify() # -- word area dispatch --------------------------------------------------- def read_words(self, address: int, count: int) -> list[int] | ExceptionResponse: if not 1 <= count <= DIAG_READ_LIMIT: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_VALUE) try: return self._read_words_impl(address, count) except AddressError: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_ADDRESS) def _read_words_impl(self, address: int, count: int) -> list[int]: end = address + count if WORD_IN1_BASE <= address and end <= WORD_IN1_BASE + AREA1_INPUT_WORDS: return self._in_words[address - WORD_IN1_BASE:end - WORD_IN1_BASE] if WORD_OUT1_MIRROR_BASE <= address and end <= WORD_OUT1_MIRROR_BASE + AREA1_OUTPUT_WORDS: off = address - WORD_OUT1_MIRROR_BASE return self._out_words[off:off + count] if WORD_CONFIG_BASE <= address and end <= WORD_CONFIG_END + 1: return self._config.read_config(address, count) if WORD_IN2_BASE <= address and end <= WORD_IN2_BASE + AREA2_INPUT_WORDS: off = address - WORD_IN2_BASE + AREA1_INPUT_WORDS return self._in_words[off:off + count] if WORD_OUT2_MIRROR_BASE <= address and end <= WORD_OUT2_MIRROR_BASE + AREA2_OUTPUT_WORDS: off = address - WORD_OUT2_MIRROR_BASE + AREA1_OUTPUT_WORDS return self._out_words[off:off + count] raise AddressError(address) def write_words(self, address: int, values: list[int]) -> None | ExceptionResponse: if not 1 <= len(values) <= DIAG_WRITE_LIMIT: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_VALUE) try: self._write_words_impl(address, values) return None except AddressError: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_ADDRESS) def _write_words_impl(self, address: int, values: list[int]) -> None: count = len(values) end = address + count if WORD_OUT1_BASE <= address and end <= WORD_OUT1_BASE + AREA1_OUTPUT_WORDS: self._out_words[address - WORD_OUT1_BASE:end - WORD_OUT1_BASE] = [ v & 0xFFFF for v in values ] self._notify() return if WORD_OUT1_MIRROR_BASE <= address and end <= WORD_OUT1_MIRROR_BASE + AREA1_OUTPUT_WORDS: off = address - WORD_OUT1_MIRROR_BASE self._out_words[off:off + count] = [v & 0xFFFF for v in values] self._notify() return if WORD_CONFIG_BASE <= address and end <= WORD_CONFIG_END + 1: self._config.write_config(address, values) return if WORD_OUT2_BASE <= address and end <= WORD_OUT2_BASE + AREA2_OUTPUT_WORDS: off = address - WORD_OUT2_BASE + AREA1_OUTPUT_WORDS self._out_words[off:off + count] = [v & 0xFFFF for v in values] self._notify() return if WORD_OUT2_MIRROR_BASE <= address and end <= WORD_OUT2_MIRROR_BASE + AREA2_OUTPUT_WORDS: off = address - WORD_OUT2_MIRROR_BASE + AREA1_OUTPUT_WORDS self._out_words[off:off + count] = [v & 0xFFFF for v in values] self._notify() return raise AddressError(address) def mask_write_word(self, address: int, and_mask: int, or_mask: int) -> int | ExceptionResponse: """FC22: result = (current AND and_mask) OR (or_mask AND NOT and_mask). The 'current' value is the register under the WRITE mapping (FC6/FC16, Table 120): address 0x0000-0x00FF is %QW0-255. """ try: current = self._read_write_target(address) except AddressError: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_ADDRESS) value = (current & and_mask) | (or_mask & ~and_mask & 0xFFFF) result = self.write_words(address, [value]) if isinstance(result, ExceptionResponse): return result return value def _read_write_target(self, address: int) -> int: """Current value at `address` under the write (FC6/FC16) mapping.""" if WORD_OUT1_BASE <= address < WORD_OUT1_BASE + AREA1_OUTPUT_WORDS: return self._out_words[address - WORD_OUT1_BASE] if (WORD_OUT1_MIRROR_BASE <= address < WORD_OUT1_MIRROR_BASE + AREA1_OUTPUT_WORDS): return self._out_words[address - WORD_OUT1_MIRROR_BASE] if WORD_CONFIG_BASE <= address <= WORD_CONFIG_END: return self._config.read_config(address, 1)[0] if WORD_OUT2_BASE <= address < WORD_OUT2_BASE + AREA2_OUTPUT_WORDS: return self._out_words[address - WORD_OUT2_BASE + AREA1_OUTPUT_WORDS] if (WORD_OUT2_MIRROR_BASE <= address < WORD_OUT2_MIRROR_BASE + AREA2_OUTPUT_WORDS): return self._out_words[ address - WORD_OUT2_MIRROR_BASE + AREA1_OUTPUT_WORDS ] raise AddressError(address) # -- bit area dispatch ----------------------------------------------------- def read_bits(self, address: int, count: int) -> list[bool] | ExceptionResponse: """FC1/FC2 bit reads: the ADDRESS selects the area (Table 121). 0x0000-0x01FF -> inputs (channels 1-512), 0x0200-0x03FF -> outputs (read-back mirror), 0x8000-0x85F7 -> inputs 513-2039, 0x9000-0x95F7 -> outputs 513-2039. """ if not 1 <= count <= COIL_READ_LIMIT: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_VALUE) try: return self._read_bits_impl(address, count) except AddressError: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_ADDRESS) def _read_bits_impl(self, address: int, count: int) -> list[bool]: direction, g0, area_bits = self._bit_read_area(address) if g0 + count > area_bits: raise AddressError(address) # spans into an illegal area words, offset = self._direction_arrays(direction) bits: list[bool] = [] for i in range(count): g = g0 + i word = offset + g // 16 if word >= len(words): raise AddressError(address) # beyond configured image bits.append(bool(words[word] & (1 << (g % 16)))) return bits def write_bits(self, address: int, values: list[bool]) -> None | ExceptionResponse: if not 1 <= len(values) <= COIL_WRITE_LIMIT: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_VALUE) try: self._write_bits_impl(address, values) return None except AddressError: return ExceptionResponse(0x03, EXC_ILLEGAL_DATA_ADDRESS) def _write_bits_impl(self, address: int, values: list[bool]) -> None: direction, g0, area_bits = self._bit_write_area(address) if g0 + len(values) > area_bits: raise AddressError(address) # spans into an illegal area words, offset = self._direction_arrays(direction) for i, value in enumerate(values): g = g0 + i word = offset + g // 16 if word >= len(words): raise AddressError(address) # beyond configured image if value: words[word] |= 1 << (g % 16) else: words[word] &= ~(1 << (g % 16)) & 0xFFFF self._notify() def _direction_arrays(self, direction: str) -> tuple[list[int], int]: if direction == "in": return self._in_words, self._digital_in_word_offset return self._out_words, self._digital_out_word_offset def _bit_read_area(self, address: int) -> tuple[str, int, int]: """Map a bit-read address to (direction, global bit index, area size). The global bit index is the digital channel number - 1; storage word = digital offset + g//16 (byte-packing, §7.3.2.1). """ if BIT_IN1_BASE <= address < BIT_IN1_BASE + BIT_IN1_COUNT: return "in", address, BIT_IN1_COUNT if BIT_OUT1_BASE <= address < BIT_OUT1_BASE + BIT_OUT1_COUNT: return "out", address - BIT_OUT1_BASE, BIT_OUT1_COUNT if BIT_IN2_BASE <= address < BIT_IN2_BASE + BIT_IN2_COUNT: return "in", BIT_IN1_COUNT + (address - BIT_IN2_BASE), BIT_IN2_COUNT if BIT_OUT2_BASE <= address < BIT_OUT2_BASE + BIT_OUT2_COUNT: return "out", BIT_OUT1_COUNT + (address - BIT_OUT2_BASE), BIT_OUT2_COUNT raise AddressError(address) def _bit_write_area(self, address: int) -> tuple[str, int, int]: """Map a bit-write address to (direction, global bit index, area size). Table 122: ALL bit writes target OUTPUT areas — 0x0000-0x01FF and 0x0200-0x03FF -> outputs 1-512, 0x8000-0x85F7 and 0x9000-0x95F7 -> outputs 513-2039. """ if address < BIT_IN2_BASE: if address < BIT_IN1_COUNT: return "out", address, BIT_IN1_COUNT if BIT_OUT1_BASE <= address < BIT_OUT1_BASE + BIT_OUT1_COUNT: return "out", address - BIT_OUT1_BASE, BIT_OUT1_COUNT raise AddressError(address) if BIT_IN2_BASE <= address < BIT_IN2_BASE + BIT_IN2_COUNT: return "out", BIT_IN1_COUNT + (address - BIT_IN2_BASE), BIT_IN2_COUNT if BIT_OUT2_BASE <= address < BIT_OUT2_BASE + BIT_OUT2_COUNT: return "out", BIT_OUT1_COUNT + (address - BIT_OUT2_BASE), BIT_OUT2_COUNT raise AddressError(address) # -- helpers for the scenario engine --------------------------------------- def digital_input_channel(self, channel: int) -> bool: """Digital input channel number (1-based) -> state.""" return self._bit_value( self._in_words, self._digital_in_word_offset, channel - 1 ) def set_digital_input_channel(self, channel: int, value: bool) -> None: self._set_bit( self._in_words, self._digital_in_word_offset, channel - 1, value ) def digital_output_channel(self, channel: int) -> bool: return self._bit_value( self._out_words, self._digital_out_word_offset, channel - 1 ) @staticmethod def _bit_value(words: list[int], offset: int, bit_index: int) -> bool: return bool(words[offset + bit_index // 16] & (1 << (bit_index % 16))) @staticmethod def _set_bit(words: list[int], offset: int, bit_index: int, value: bool) -> None: w = offset + bit_index // 16 if value: words[w] = words[w] | (1 << (bit_index % 16)) else: words[w] = words[w] & ~(1 << (bit_index % 16)) & 0xFFFF @staticmethod def _validate_index(index: int, size: int) -> None: if not 0 <= index < size: raise IndexError(f"word index out of range: {index}")