"""Configuration, diagnostic, constant, and firmware-info register bank. Implements the 0x1000-0x2FFF area of the WAGO 750-362 (manual ยง11.2.5). Watchdog registers 0x1000-0x100B are delegated to :class:`ModbusWatchdog`. """ from __future__ import annotations from collections.abc import Callable from .stats import STATS_REGISTER, STATS_WORD_COUNT, ModbusStatistics from .watchdog import ModbusWatchdog, RegisterWriteRefused # --- register addresses ------------------------------------------------------- REG_LED_ERROR_CODE = 0x1020 REG_LED_ERROR_ARG = 0x1021 REG_ANALOG_OUT_BITS = 0x1022 REG_ANALOG_IN_BITS = 0x1023 REG_DIGITAL_OUT_BITS = 0x1024 REG_DIGITAL_IN_BITS = 0x1025 REG_TCP_CONNECTIONS = 0x102A REG_KBUS_RESET = 0x102B REG_TCP_IDLE_TIMEOUT = 0x1030 REG_MAC_ID = 0x1031 REG_RESPONSE_DELAY = 0x1037 REG_TOS = 0x1038 REG_MODULE_DIAG = 0x1050 # constants 0x2000-0x2008 CONSTANTS = { 0x2000: 0x0000, 0x2001: 0xFFFF, 0x2002: 0x1234, 0x2003: 0xAAAA, 0x2004: 0x5555, 0x2005: 0x7FFF, 0x2006: 0x8000, 0x2007: 0x3FFF, 0x2008: 0x4000, } # firmware info 0x2010-0x2043 REG_FW_REVISION = 0x2010 REG_FW_SERIES = 0x2011 REG_FW_ITEM = 0x2012 REG_FW_MAJOR = 0x2013 REG_FW_MINOR = 0x2014 REG_FW_DESCRIPTION = 0x2020 # up to 16 words REG_FW_COMPILE_TIME = 0x2021 # up to 8 words REG_FW_COMPILE_DATE = 0x2022 # up to 8 words REG_FW_LOADER = 0x2023 # up to 32 words REG_MODULE_DESC_0_64 = 0x2030 # 65 words REG_PROCESS_IMAGE_SETTINGS = 0x2035 REG_DIAGNOSTICS = 0x2036 REG_SOFTWARE_RESET = 0x2040 # write 0x55AA/0xAA55 REG_FACTORY_SETTINGS = 0x2043 RESET_KEYS = (0x55AA, 0xAA55) SERIES_CODE_750 = 0x02EE # 750 dec COUPLER_CODE_362 = 0x016A # 362 dec def _string_to_words(text: str, max_words: int) -> list[int]: """Pack ASCII into 16-bit words (2 chars/word, high byte first).""" data = text.encode("ascii", errors="replace")[: 2 * max_words] data = data.ljust(2 * max_words, b"\x00") return [int.from_bytes(data[2 * i : 2 * i + 2], "big") for i in range(max_words)] class ConfigRegisterBank: """Read/write dispatch for the 0x1000-0x2FFF configuration area.""" def __init__( self, *, statistics: ModbusStatistics, watchdog: ModbusWatchdog, tcp_connection_counter: Callable[[], int], mac_id: tuple[int, int, int, int, int, int] = (0x00, 0x30, 0xDE, 0x11, 0x22, 0x33), response_delay_ms: int = 0, idle_timeout_units: int = 600, process_image_bits: tuple[int, int, int, int] = (0, 0, 0, 0), firmware_description: str = "WAGO 750-362 Simulator", on_software_reset: Callable[[], None] | None = None, on_kbus_reset: Callable[[], None] | None = None, ) -> None: # (analog_in_bits, analog_out_bits, digital_in_bits, digital_out_bits) self._process_image_bits = process_image_bits self._stats = statistics self._watchdog = watchdog self._tcp_connections = tcp_connection_counter self._mac_id = mac_id self._response_delay_ms = response_delay_ms self._idle_timeout_units = idle_timeout_units self._led_error_code = 0 self._led_error_arg = 0 self._fw = { REG_FW_REVISION: 0x0001, REG_FW_SERIES: SERIES_CODE_750, REG_FW_ITEM: COUPLER_CODE_362, REG_FW_MAJOR: 0x0000, REG_FW_MINOR: 0x0001, } self._description = _string_to_words(firmware_description, 16) self._compile_time = _string_to_words("10:00:00", 8) self._compile_date = _string_to_words("2026-10-03", 8) self._loader = _string_to_words("boot", 32) self._module_desc: list[int] = [0] * 65 self._module_diag: tuple[int, int, int] = (0, 0, 0) # module, channel, diag self._on_software_reset = on_software_reset self._on_kbus_reset = on_kbus_reset # -- image geometry (set by the node after module layout) ------------------- def set_process_image_bits( self, analog_in: int, analog_out: int, digital_in: int, digital_out: int ) -> None: self._process_image_bits = (analog_in, analog_out, digital_in, digital_out) def set_module_description(self, text: str) -> None: self._module_desc = _string_to_words(text, 65) def set_led_error(self, code: int, argument: int = 0) -> None: self._led_error_code = code & 0xFFFF self._led_error_arg = argument & 0xFFFF def set_module_diagnosis(self, module: int, channel: int, diagnosis: int) -> None: """Fault injection for register 0x1050.""" self._module_diag = (module & 0xFFFF, channel & 0xFFFF, diagnosis & 0xFFFF) @property def response_delay_ms(self) -> int: return self._response_delay_ms @property def idle_timeout_ms(self) -> int: return self._idle_timeout_units * 100 # -- register access ----------------------------------------------------------- def read_config(self, address: int, count: int) -> list[int]: """Read `count` words starting at `address`. The 9-word statistics block (0x1029-0x1031) overlaps dedicated registers in the manual (0x102A connections, 0x1031 MAC-ID): a read *starting at* 0x1029 returns the statistics block as a unit; individual addresses keep their dedicated meanings. """ out: list[int] = [] if address <= STATS_REGISTER and address + count > STATS_REGISTER: out.extend(self._read_one(a) for a in range(address, STATS_REGISTER)) n_stats = min(count - len(out), STATS_WORD_COUNT) out.extend(self._stats.as_words()[:n_stats]) rest_at = STATS_REGISTER + n_stats out.extend(self._read_one(a) for a in range(rest_at, address + count)) return out return [self._read_one(a) for a in range(address, address + count)] def write_config(self, address: int, values: list[int]) -> None: for i, value in enumerate(values): self._write_one(address + i, value) def _read_one(self, address: int) -> int: # watchdog block if 0x1000 <= address <= 0x100B: return self._watchdog.read(address) # diagnostics if address == REG_LED_ERROR_CODE: return self._led_error_code if address == REG_LED_ERROR_ARG: return self._led_error_arg # image sizes if address == REG_ANALOG_IN_BITS: return self._process_image_bits[0] & 0xFFFF if address == REG_ANALOG_OUT_BITS: return self._process_image_bits[1] & 0xFFFF if address == REG_DIGITAL_IN_BITS: return self._process_image_bits[2] & 0xFFFF if address == REG_DIGITAL_OUT_BITS: return self._process_image_bits[3] & 0xFFFF # statistics if address == STATS_REGISTER: return self._stats.slave_device_failure # first word if address == REG_TCP_CONNECTIONS: return self._tcp_connections() if address == REG_KBUS_RESET: return 0 if address == REG_TCP_IDLE_TIMEOUT: return self._idle_timeout_units if REG_MAC_ID <= address < REG_MAC_ID + 3: return self._mac_id_word(address - REG_MAC_ID) if address == REG_RESPONSE_DELAY: return self._response_delay_ms if address == REG_TOS: return 0 if address == REG_MODULE_DIAG: return self._module_diag[0] if REG_MODULE_DIAG < address < REG_MODULE_DIAG + 3: return self._module_diag[address - REG_MODULE_DIAG] # constants if address in CONSTANTS: return CONSTANTS[address] # firmware info if address in self._fw: return self._fw[address] if REG_FW_DESCRIPTION <= address < REG_FW_DESCRIPTION + 16: return self._description[address - REG_FW_DESCRIPTION] if REG_FW_COMPILE_TIME <= address < REG_FW_COMPILE_TIME + 8: return self._compile_time[address - REG_FW_COMPILE_TIME] if REG_FW_COMPILE_DATE <= address < REG_FW_COMPILE_DATE + 8: return self._compile_date[address - REG_FW_COMPILE_DATE] if REG_FW_LOADER <= address < REG_FW_LOADER + 32: return self._loader[address - REG_FW_LOADER] if REG_MODULE_DESC_0_64 <= address < REG_MODULE_DESC_0_64 + 65: return self._module_desc[address - REG_MODULE_DESC_0_64] if address == REG_PROCESS_IMAGE_SETTINGS: return 0 if address == REG_DIAGNOSTICS: return 0 # undefined registers within the config area read as 0 return 0 def _mac_id_word(self, index: int) -> int: """Word i of the MAC-ID register (0x1031, 3 words). The register exposes the first 3 MAC bytes as words (one byte per word, per the manual's 3-word layout for 6-byte MACs). """ return self._mac_id[index] << 8 def _write_one(self, address: int, value: int) -> None: value &= 0xFFFF if 0x1000 <= address <= 0x100B: self._watchdog.write(address, value) return if address == STATS_REGISTER: self._stats.handle_reset_write(value) return if address == REG_TCP_IDLE_TIMEOUT: self._idle_timeout_units = value return if address == REG_RESPONSE_DELAY: self._response_delay_ms = value return if address == REG_TOS: return # stored; echo on read is 0 in v1 if address == REG_KBUS_RESET: if value in RESET_KEYS: if self._on_kbus_reset: self._on_kbus_reset() return if address == REG_SOFTWARE_RESET: if value in RESET_KEYS: if self._on_software_reset: self._on_software_reset() return if address == REG_FACTORY_SETTINGS: return # remaining registers are read-only raise RegisterWriteRefused(f"register {address:#06x} is read-only")