#!/usr/bin/env python3 """Pure-Python MSCB client using the Ethernet submaster UDP protocol. The module intentionally exposes the traditional ``mscb_*`` functions so existing Python callers can keep their command and status handling. RPC and USB transports are not present. """ from __future__ import annotations import ctypes import socket import struct import threading import time from dataclasses import dataclass, field from pathlib import Path from types import SimpleNamespace from typing import Any, Optional MSCB_LIBRARY_VERSION = "2.7.0-py" MSCB_PROTOCOL_VERSION = 5 MSCB_NET_PORT = 1177 MSCB_MAX_FD = 1024 # MSCB commands MCMD_ADDR_NODE8 = 0x09 MCMD_ADDR_NODE16 = 0x0A MCMD_ADDR_BC = 0x10 MCMD_ADDR_GRP8 = 0x11 MCMD_ADDR_GRP16 = 0x12 MCMD_PING8 = 0x19 MCMD_PING16 = 0x1A MCMD_INIT = 0x20 MCMD_GET_INFO = 0x28 MCMD_SET_ADDR = 0x33 MCMD_SET_NAME = 0x37 MCMD_SET_BAUD = 0x39 MCMD_FREEZE = 0x41 MCMD_SET_TIME = 0x4E MCMD_UPGRADE = 0x50 MCMD_USER = 0x58 MCMD_ECHO = 0x61 MCMD_TOKEN = 0x68 MCMD_GET_UPTIME = 0x70 MCMD_ACK = 0x78 MCMD_WRITE_NA = 0x80 MCMD_WRITE_ACK = 0x88 MCMD_FLASH = 0x98 MCMD_READ = 0xA0 MCMD_WRITE_RANGE = 0xA8 MCMD_WRITE_MEM = 0xB7 MCMD_READ_MEM = 0xBF MCMD_LOG = 0xC1 ADDR_SET_NODE = 1 ADDR_SET_HIGH = 2 ADDR_SET_GROUP = 3 UCMD_ECHO = 1 UCMD_ERASE = 2 UCMD_PROGRAM = 3 UCMD_VERIFY = 4 UCMD_READ = 5 UCMD_REBOOT = 6 UCMD_RETURN = 7 # Submaster RS485 flags RS485_FLAG_BIT9 = 1 << 0 RS485_FLAG_NO_ACK = 1 << 1 RS485_FLAG_SHORT_TO = 1 << 2 RS485_FLAG_LONG_TO = 1 << 3 RS485_FLAG_CMD = 1 << 4 RS485_FLAG_ADR_CYCLE = 1 << 5 RS485_FLAG_NO_RETRY = 1 << 6 RS485_FLAG_VERYLONG_TO = 1 << 7 # Status and error codes MSCB_SUCCESS = 1 MSCB_CRC_ERROR = 2 MSCB_TIMEOUT = 3 MSCB_TIMEOUT_BUS = 4 MSCB_INVAL_PARAM = 5 MSCB_MUTEX = 6 MSCB_FORMAT_ERROR = 7 MSCB_NO_MEM = 8 MSCB_SUBM_ERROR = 9 MSCB_ADDR_EXISTS = 10 MSCB_WRONG_PASS = 11 MSCB_SUBADDR = 12 MSCB_NOTREADY = 13 MSCB_NO_VAR = 14 MSCB_INVALID_INDEX = 15 MSCB_NOT_FOUND = 16 EMSCB_UNDEFINED = -1 EMSCB_NO_MEM = -2 EMSCB_NO_ACCESS = -4 EMSCB_LOCKED = -5 EMSCB_NO_SUBM = -6 EMSCB_INVAL_PARAM = -7 EMSCB_WRONG_PASSWORD = -8 EMSCB_COMM_ERROR = -9 EMSCB_NOT_FOUND = -10 EMSCB_NO_WRITE_ACCESS = -11 EMSCB_PROTOCOL_VERSION = -12 MSCB_UPLOAD_DEBUG = 0x01 MSCB_UPLOAD_SUBADDR = 0x02 MSCB_BASE_RAM = 0x00000000 MSCB_BASE_NVRAM = 0x10000000 MSCB_BASE_FLASH = 0x20000000 MSCB_BASE_CODE = 0x30000000 XIL_FW_ADDR = 0x00000000 XIL_SW_ADDR = 0x00420000 XIL_FW_HEADER_ADDR = XIL_SW_ADDR - 0x1000 XIL_SW_HEADER_ADDR = XIL_SW_ADDR + 0x001FF000 XIL_BITFILE_HEADER_ID = 0xBF8EADE9 _HEADER = struct.Struct("!HHBB") _debug_flag = 0 _max_retry = 10 class MSCB_INFO(ctypes.Structure): _fields_ = [ ("protocol_version", ctypes.c_ubyte), ("n_variables", ctypes.c_ubyte), ("node_address", ctypes.c_ushort), ("group_address", ctypes.c_ushort), ("revision", ctypes.c_ushort), ("node_name", ctypes.c_char * 16), ("rtc", ctypes.c_ubyte * 6), ("buf_size", ctypes.c_ushort), ("pinExt_resets", ctypes.c_ushort), ("SW_resets", ctypes.c_ushort), ("int_WD_resets", ctypes.c_ushort), ("bootBank", ctypes.c_ubyte), ("siliconRevison", ctypes.c_ubyte), ("systemLoad", ctypes.c_float), ("peakSystemLoad", ctypes.c_float), ("pcbTemp", ctypes.c_float), ("Supply1V8", ctypes.c_float), ("Supply3V3", ctypes.c_float), ("Supply5V0", ctypes.c_float), ("Supply24V0", ctypes.c_float), ("Supply5V0Ext", ctypes.c_float), ("Supply24V0Ext", ctypes.c_float), ("Supply24V0Current", ctypes.c_float), ("Vbat", ctypes.c_float), ] class MSCB_INFO_VAR(ctypes.Structure): _fields_ = [ ("width", ctypes.c_ubyte), ("unit", ctypes.c_ubyte), ("prefix", ctypes.c_ubyte), ("status", ctypes.c_ubyte), ("flags", ctypes.c_ubyte), ("name", ctypes.c_char * 16), ] def _debug_log(message: str, write: bool = False) -> None: if not _debug_flag or (_debug_flag == 1 and not write): return line = f"{time.strftime('%H:%M:%S')} {threading.current_thread().name} {message}" if _debug_flag == 3: print(line) return filename = "mscb_write.log" if _debug_flag == 1 else "mscb_debug.log" try: with open(filename, "a", encoding="utf-8") as stream: stream.write(line + "\n") except OSError: pass def crc8(data: bytes | bytearray | memoryview, length: Optional[int] = None) -> int: """Return the Dallas/Maxim CRC-8 used by MSCB.""" value = 0 view = memoryview(data) if length is not None: view = view[:length] for octet in view: value ^= octet for _ in range(8): value = ((value >> 1) ^ 0x8C) if value & 1 else value >> 1 return value def _as_int(value: Any) -> int: return int(value.value) if hasattr(value, "value") else int(value) def _as_bytes(value: Any, size: Optional[int] = None) -> bytes: if value is None: return b"" if isinstance(value, str): raw = value.encode("latin-1") elif isinstance(value, (bytes, bytearray, memoryview)): raw = bytes(value) elif hasattr(value, "value") and isinstance(value.value, bytes): raw = value.value else: if size is None: size = ctypes.sizeof(value) raw = ctypes.string_at(value, size) return raw if size is None else raw[:size] def _set_scalar(pointer: Any, ctype: Any, value: int) -> None: ctypes.cast(pointer, ctypes.POINTER(ctype))[0] = value def _get_scalar(pointer: Any, ctype: Any) -> int: return int(ctypes.cast(pointer, ctypes.POINTER(ctype))[0]) def _copy_to_buffer(buffer: Any, payload: bytes, capacity: Optional[int] = None) -> None: if buffer is None: return if capacity is not None and capacity > 0: ctypes.memset(buffer, 0, capacity) payload = payload[:capacity] if payload: ctypes.memmove(buffer, payload, len(payload)) def _decode_text(value: Any) -> str: if isinstance(value, str): return value raw = _as_bytes(value) return raw.split(b"\0", 1)[0].decode("latin-1", errors="ignore") def _target(pointer: Any) -> Any: return getattr(pointer, "_obj", pointer) def _node_address(adr: int, command: int = MCMD_ADDR_NODE16) -> bytes: part = bytes((command, (adr >> 8) & 0xFF, adr & 0xFF)) return part + bytes((crc8(part),)) @dataclass class _Connection: device: str host: str port: int sock: socket.socket seq: int = 0 eth_max_retry: int = 10 eth_pause_ms: int = 0 last_comm: float = 0.0 lock: threading.RLock = field(default_factory=threading.RLock) info_cache: dict[tuple[int, int], Optional[bytes]] = field(default_factory=dict) def exchange(self, request: bytes, flags: int, capacity: int = 1500) -> tuple[int, bytes]: if not 1 <= len(request) < 1500: return MSCB_INVAL_PARAM, b"" with self.lock: if self.eth_pause_ms: wait = self.eth_pause_ms / 1000.0 - (time.monotonic() - self.last_comm) if wait > 0: time.sleep(wait) self.last_comm = time.monotonic() retries = max(1, self.eth_max_retry) last_status = MSCB_TIMEOUT for retry in range(retries): self.seq = (self.seq + 1) & 0xFFFF packet = _HEADER.pack(len(request), self.seq, flags & 0xFF, MSCB_PROTOCOL_VERSION) + request _debug_log(f"udp send seq={self.seq} flags=0x{flags:02x} data={request.hex()}") try: if self.sock.sendto(packet, (self.host, self.port)) != len(packet): return MSCB_TIMEOUT, b"" except OSError: return MSCB_TIMEOUT, b"" if flags & RS485_FLAG_NO_ACK: return MSCB_SUCCESS, b"" timeout_ms = 1000 if flags & RS485_FLAG_LONG_TO else 300 * (retry + 1) if flags & RS485_FLAG_VERYLONG_TO and retry > 0: timeout_ms = 5000 last_status, response = self._receive(self.seq, timeout_ms / 1000.0, capacity) if last_status == EMSCB_PROTOCOL_VERSION: return last_status, b"" if response: _debug_log(f"udp recv seq={self.seq} data={response.hex()}") # A single 0xff is a timeout reported by the remote RS485 bus. return MSCB_SUCCESS, response if flags & RS485_FLAG_NO_RETRY: break if flags & RS485_FLAG_VERYLONG_TO and retry > 1: break return last_status, b"" def _receive(self, sequence: int, timeout: float, capacity: int) -> tuple[int, bytes]: deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: return MSCB_TIMEOUT, b"" self.sock.settimeout(remaining) try: packet, _ = self.sock.recvfrom(1500) except socket.timeout: return MSCB_TIMEOUT, b"" except OSError: return MSCB_TIMEOUT, b"" if len(packet) < _HEADER.size: continue size, received_seq, _flags, version = _HEADER.unpack_from(packet) if version != MSCB_PROTOCOL_VERSION: return EMSCB_PROTOCOL_VERSION, b"" if size + _HEADER.size != len(packet): continue if received_seq != sequence: continue if size > capacity: return MSCB_INVAL_PARAM, b"" return MSCB_SUCCESS, packet[_HEADER.size:] _connections: dict[int, _Connection] = {} _connections_lock = threading.Lock() _link_cache: dict[tuple[int, int, int], tuple[bytes, int, float]] = {} def _connection(fd: int) -> Optional[_Connection]: return _connections.get(_as_int(fd)) def _exchange(fd: int, request: bytes, flags: int, capacity: int = 1500) -> tuple[int, bytes]: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM, b"" return conn.exchange(request, flags, capacity) def mscb_exchg(fd: int, buffer: Any, size: Any, length: int, flags: int) -> int: """C-compatible low-level exchange entry point.""" capacity = _get_scalar(size, ctypes.c_int) if size is not None else 1500 status, response = _exchange(fd, _as_bytes(buffer, _as_int(length)), _as_int(flags), capacity) if size is not None: _copy_to_buffer(buffer, response, capacity) _set_scalar(size, ctypes.c_int, len(response)) return status def mscb_init(device: Any, device_size: int = 0, password: Any = b"", debug: int = 0) -> int: """Open and authenticate an Ethernet MSCB submaster.""" del device_size name = _decode_text(device) if not name: return EMSCB_INVAL_PARAM if name.count(":") > 1: return EMSCB_INVAL_PARAM # The Ethernet transport accepts IPv4 host[:port]. host, separator, port_text = name.partition(":") try: port = int(port_text) if separator else MSCB_NET_PORT socket.gethostbyname(host) except (OSError, ValueError): return EMSCB_NOT_FOUND with _connections_lock: for fd, existing in _connections.items(): if existing.device == name: return fd if len(_connections) >= MSCB_MAX_FD: return EMSCB_NO_MEM fd = next(number for number in range(1, MSCB_MAX_FD + 1) if number not in _connections) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) conn = _Connection(name, host, port, sock) _connections[fd] = conn _link_cache.clear() mscb_debug(_as_int(debug)) status, response = conn.exchange(bytes((MCMD_ECHO,)), RS485_FLAG_CMD | RS485_FLAG_VERYLONG_TO, 64) if status == EMSCB_PROTOCOL_VERSION: mscb_exit(fd) return EMSCB_PROTOCOL_VERSION if status != MSCB_SUCCESS or len(response) < 2: mscb_exit(fd) return EMSCB_COMM_ERROR if len(response) < 4 or response[1] != MSCB_PROTOCOL_VERSION: mscb_exit(fd) return EMSCB_PROTOCOL_VERSION token = _as_bytes(password).split(b"\0", 1)[0][:19] request = bytes((MCMD_TOKEN,)) + token + b"\0" * (20 - len(token)) status, response = conn.exchange(request, RS485_FLAG_CMD, 64) if status != MSCB_SUCCESS or len(response) != 1: mscb_exit(fd) return EMSCB_COMM_ERROR if response[0] == 0xFF: mscb_exit(fd) return EMSCB_WRONG_PASSWORD if response[0] != MCMD_ACK: mscb_exit(fd) return EMSCB_COMM_ERROR return fd def mscb_exit(fd: int) -> int: with _connections_lock: conn = _connections.pop(_as_int(fd), None) if conn is None: return MSCB_INVAL_PARAM conn.sock.close() return MSCB_SUCCESS def mscb_debug(flag: int) -> int: global _debug_flag _debug_flag = _as_int(flag) return MSCB_SUCCESS def mscb_get_version(lib_version: Any, protocol_version: Any) -> None: _copy_to_buffer(lib_version, MSCB_LIBRARY_VERSION.encode() + b"\0") _copy_to_buffer(protocol_version, str(MSCB_PROTOCOL_VERSION).encode() + b"\0") def mscb_get_device(fd: int, device: Any, bufsize: int) -> None: conn = _connection(fd) text = (conn.device.encode() + b"\0") if conn else b"\0" _copy_to_buffer(device, text, _as_int(bufsize)) def mscb_get_max_retry() -> int: return _max_retry def mscb_set_max_retry(max_retry: int) -> int: global _max_retry old = _max_retry _max_retry = max(1, _as_int(max_retry)) return old def mscb_get_eth_max_retry(fd: int) -> int: conn = _connection(fd) return conn.eth_max_retry if conn else MSCB_INVAL_PARAM def mscb_set_eth_max_retry(fd: int, max_retry: int) -> int: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM old = conn.eth_max_retry conn.eth_max_retry = max(1, _as_int(max_retry)) return old def mscb_get_eth_pause(fd: int) -> int: conn = _connection(fd) return conn.eth_pause_ms if conn else MSCB_INVAL_PARAM def mscb_set_eth_pause(fd: int, pause: int) -> int: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM conn.eth_pause_ms = max(0, _as_int(pause)) return MSCB_SUCCESS def mscb_addr(fd: int, command: int, adr: int, quick: int, retry: int) -> int: command, adr = _as_int(command), _as_int(adr) & 0xFFFF attempts = max(1, _as_int(retry)) for _ in range(attempts): if command in (MCMD_ADDR_NODE8, MCMD_ADDR_GRP8, MCMD_PING8): body = bytes((command, adr & 0xFF)) elif command in (MCMD_ADDR_NODE16, MCMD_ADDR_GRP16, MCMD_PING16): body = bytes((command, adr >> 8, adr & 0xFF)) else: body = bytes((command,)) request = body + bytes((crc8(body),)) flags = RS485_FLAG_BIT9 | RS485_FLAG_SHORT_TO if command not in (MCMD_PING8, MCMD_PING16): flags |= RS485_FLAG_NO_ACK elif not _as_int(quick): flags &= ~RS485_FLAG_SHORT_TO status, response = _exchange(fd, request, flags, 64) if status != MSCB_SUCCESS: return MSCB_SUBM_ERROR if command not in (MCMD_PING8, MCMD_PING16): return MSCB_SUCCESS if response and response[0] == MCMD_ACK: return MSCB_SUCCESS if attempts > 1: _exchange(fd, b"\0" * 10, RS485_FLAG_BIT9 | RS485_FLAG_NO_ACK) time.sleep(0.01) return MSCB_TIMEOUT def mscb_ping(fd: int, adr: int, quick: int = 0, retry: int = 0) -> int: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM attempts = conn.eth_max_retry if _as_int(retry) else 1 return mscb_addr(fd, MCMD_PING16, adr, quick, attempts) def _request_node(fd: int, adr: int, command: bytes, capacity: int = 256, flags: int = RS485_FLAG_ADR_CYCLE) -> tuple[int, bytes]: return _exchange(fd, _node_address(_as_int(adr)) + command, flags, capacity) def _read_node_info(fd: int, adr: int) -> tuple[int, Optional[SimpleNamespace]]: command = bytes((MCMD_GET_INFO, crc8(bytes((MCMD_GET_INFO,))))) response = b"" for _ in range(_max_retry): status, response = _request_node(fd, adr, command) if status == MSCB_SUCCESS and len(response) >= 27: break if len(response) < 27: return MSCB_TIMEOUT, None if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR, None raw = response[2:-1] padded = raw + b"\0" * max(0, 84 - len(raw)) values = SimpleNamespace( protocol_version=padded[0], n_variables=padded[1], node_address=int.from_bytes(padded[2:4], "big"), group_address=int.from_bytes(padded[4:6], "big"), revision=int.from_bytes(padded[6:8], "big"), node_name=padded[8:24], rtc=padded[24:30], buf_size=int.from_bytes(padded[30:32], "big"), pinExt_resets=int.from_bytes(padded[32:34], "big"), SW_resets=int.from_bytes(padded[34:36], "big"), int_WD_resets=int.from_bytes(padded[36:38], "big"), bootBank=padded[38], siliconRevison=padded[39], ) float_names = ("systemLoad", "peakSystemLoad", "pcbTemp", "Supply1V8", "Supply3V3", "Supply5V0", "Supply24V0", "Supply5V0Ext", "Supply24V0Ext", "Supply24V0Current", "Vbat") for offset, name in enumerate(float_names, 40): pos = 40 + (offset - 40) * 4 setattr(values, name, struct.unpack(">f", padded[pos:pos + 4])[0]) return MSCB_SUCCESS, values def mscb_info(fd: int, adr: int, info: Any) -> int: status, values = _read_node_info(fd, _as_int(adr)) if status != MSCB_SUCCESS or values is None: return status out = _target(info) for name, value in vars(values).items(): if name == "node_name": out.node_name = value[:16] elif name == "rtc": for index, octet in enumerate(value[:6]): out.rtc[index] = octet elif hasattr(out, name): setattr(out, name, value) return MSCB_SUCCESS def mscb_uptime(fd: int, adr: int, uptime: Any) -> int: body = bytes((MCMD_GET_UPTIME,)) status, response = _request_node(fd, adr, body + bytes((crc8(body),))) if status != MSCB_SUCCESS: return status if len(response) < 6: return MSCB_TIMEOUT if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR _set_scalar(uptime, ctypes.c_uint, int.from_bytes(response[1:5], "big")) return MSCB_SUCCESS def mscb_info_variable(fd: int, adr: int, index: int, info: Any) -> int: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM key = (_as_int(adr), _as_int(index) & 0xFF) if key not in conn.info_cache: body = bytes((MCMD_GET_INFO + 1, key[1])) response = b"" for _ in range(2): status, response = _request_node(fd, key[0], body + bytes((crc8(body),)), 80) if status == MSCB_SUCCESS and len(response) in (16, 24): break if len(response) < 16: if len(response) == 2: conn.info_cache[key] = None return MSCB_NO_VAR if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR conn.info_cache[key] = response[2:-1] raw = conn.info_cache[key] if raw is None: return MSCB_NO_VAR raw = raw + b"\0" * (21 - len(raw)) out = _target(info) out.width, out.unit, out.prefix, out.status, out.flags = raw[:5] out.name = raw[5:21] return MSCB_SUCCESS def mscb_clear_info_cache() -> None: for conn in _connections.values(): conn.info_cache.clear() def _selected_command(addr: int, gaddr: int, broadcast: int, command: bytes) -> Optional[bytes]: if _as_int(addr) >= 0: return _node_address(_as_int(addr), MCMD_ADDR_NODE16) + command if _as_int(gaddr) >= 0: return _node_address(_as_int(gaddr), MCMD_ADDR_GRP16) + command if _as_int(broadcast): address = bytes((MCMD_ADDR_BC, crc8(bytes((MCMD_ADDR_BC,))))) return address + command return None def _send_selected(fd: int, addr: int, gaddr: int, broadcast: int, body: bytes, no_ack: bool = True, long_timeout: bool = False) -> int: command = body + bytes((crc8(body),)) request = _selected_command(addr, gaddr, broadcast, command) if request is None: return MSCB_INVAL_PARAM flags = RS485_FLAG_ADR_CYCLE if no_ack: flags |= RS485_FLAG_NO_ACK if long_timeout: flags |= RS485_FLAG_LONG_TO status, _ = _exchange(fd, request, flags) return status def mscb_reboot(fd: int, addr: int, gaddr: int, broadcast: int) -> int: status = _send_selected(fd, addr, gaddr, broadcast, bytes((MCMD_INIT,))) mscb_clear_info_cache() return status def mscb_flash(fd: int, addr: int, gaddr: int, broadcast: int) -> int: status = _send_selected(fd, addr, gaddr, broadcast, bytes((MCMD_FLASH,))) time.sleep(0.5) return status def mscb_set_node_addr(fd: int, addr: int, gaddr: int, broadcast: int, new_addr: int) -> int: if _as_int(addr) >= 0 and mscb_ping(fd, new_addr, 0, 1) == MSCB_SUCCESS: return MSCB_ADDR_EXISTS mode = ADDR_SET_NODE if _as_int(addr) >= 0 else ADDR_SET_HIGH value = _as_int(new_addr) & 0xFFFF status = _send_selected(fd, addr, gaddr, broadcast, bytes((MCMD_SET_ADDR, mode, value >> 8, value & 0xFF))) mscb_clear_info_cache() return status def mscb_set_group_addr(fd: int, addr: int, gaddr: int, broadcast: int, new_addr: int) -> int: value = _as_int(new_addr) & 0xFFFF status = _send_selected(fd, addr, gaddr, broadcast, bytes((MCMD_SET_ADDR, ADDR_SET_GROUP, value >> 8, value & 0xFF))) mscb_clear_info_cache() return status def mscb_set_name(fd: int, adr: int, name: Any) -> int: text = _as_bytes(name).split(b"\0", 1)[0][:16] if len(text) < 16: text += b"\0" body = bytes((MCMD_SET_NAME, len(text))) + text status, _ = _request_node(fd, adr, body + bytes((crc8(body),)), flags=RS485_FLAG_ADR_CYCLE | RS485_FLAG_NO_ACK) mscb_clear_info_cache() return status def mscb_set_baud(fd: int, baud: int) -> int: address = bytes((MCMD_ADDR_BC, crc8(bytes((MCMD_ADDR_BC,))))) body = bytes((MCMD_SET_BAUD, _as_int(baud) & 0xFF)) status, _ = _exchange(fd, address + body + bytes((crc8(body),)), RS485_FLAG_ADR_CYCLE | RS485_FLAG_NO_ACK) return status def _write_value(fd: int, adr: int, index: int, payload: bytes, group: bool, retries: int) -> int: if not payload or (group and len(payload) > 4): return MSCB_INVAL_PARAM address = _node_address(_as_int(adr), MCMD_ADDR_GRP16 if group else MCMD_ADDR_NODE16) if len(payload) < 6: encoded = payload[::-1] if len(payload) < 5 else payload body = bytes(((MCMD_WRITE_NA if group else MCMD_WRITE_ACK) + len(payload) + 1, _as_int(index) & 0xFF)) + encoded else: body = bytes(((MCMD_WRITE_NA if group else MCMD_WRITE_ACK) + 7, len(payload) + 1, _as_int(index) & 0xFF)) + payload checksum = crc8(address + body) if group else crc8(body) request = address + body + bytes((checksum,)) flags = RS485_FLAG_ADR_CYCLE | (RS485_FLAG_NO_ACK if group else 0) status = MSCB_TIMEOUT for _ in range(max(1, retries)): status, response = _exchange(fd, request, flags, 256) if group: return status if status == MSCB_SUCCESS and len(response) == 2 and response[0] == MCMD_ACK and response[1] == checksum: return MSCB_SUCCESS status = MSCB_TIMEOUT if len(response) == 1 else MSCB_CRC_ERROR return status def mscb_write(fd: int, adr: int, index: int, data: Any, size: int) -> int: payload = _as_bytes(data, _as_int(size)) _debug_log(f"mscb_write fd={fd} adr={adr} index={index} data={payload.hex()}", write=True) return _write_value(fd, adr, index, payload, False, _max_retry) def mscb_write_no_retries(fd: int, adr: int, index: int, data: Any, size: int) -> int: payload = _as_bytes(data, _as_int(size)) _debug_log(f"mscb_write_no_retries fd={fd} adr={adr} index={index} data={payload.hex()}", write=True) return _write_value(fd, adr, index, payload, False, 1) def mscb_write_group(fd: int, adr: int, index: int, data: Any, size: int) -> int: payload = _as_bytes(data, _as_int(size)) _debug_log(f"mscb_write_group fd={fd} adr={adr} index={index} data={payload.hex()}", write=True) return _write_value(fd, adr, index, payload, True, 1) def mscb_write_range(fd: int, adr: int, index1: int, index2: int, data: Any, size: int) -> int: payload = _as_bytes(data, _as_int(size)) _debug_log(f"mscb_write_range fd={fd} adr={adr} indices={index1}-{index2} data={payload.hex()}", write=True) if not payload or len(payload) + 10 > 256: return MSCB_INVAL_PARAM if len(payload) < 128: body = bytes((MCMD_WRITE_RANGE | 7, len(payload) + 2, _as_int(index1) & 0xFF, _as_int(index2) & 0xFF)) + payload else: count = len(payload) + 2 body = bytes((MCMD_WRITE_RANGE | 7, 0x80 | (count >> 8), count & 0xFF, _as_int(index1) & 0xFF, _as_int(index2) & 0xFF)) + payload checksum = crc8(body) status = MSCB_TIMEOUT for _ in range(_max_retry): status, response = _request_node(fd, adr, body + bytes((checksum,))) if status == MSCB_SUCCESS and response == bytes((MCMD_ACK, checksum)): return MSCB_SUCCESS return status if status != MSCB_SUCCESS else MSCB_CRC_ERROR def _read_value(fd: int, adr: int, first: int, last: Optional[int], retries: int) -> tuple[int, bytes]: fields = bytes((_as_int(first) & 0xFF,)) if last is None else bytes((_as_int(first) & 0xFF, _as_int(last) & 0xFF)) body = bytes((MCMD_READ + len(fields),)) + fields status = MSCB_TIMEOUT for attempt in range(max(1, retries)): status, response = _request_node(fd, adr, body + bytes((crc8(body),)), 1400, RS485_FLAG_ADR_CYCLE | (RS485_FLAG_LONG_TO if last is not None else 0)) if status != MSCB_SUCCESS: continue if response == bytes((MCMD_ACK,)) and attempt > 5: mscb_clear_info_cache() return MSCB_INVALID_INDEX, b"" if len(response) == 1: status = MSCB_TIMEOUT continue if len(response) < 2: status = MSCB_TIMEOUT continue if crc8(response[:-1]) != response[-1]: status = MSCB_CRC_ERROR continue if response[0] not in (MCMD_ACK + len(response) - 2, MCMD_ACK + 7): status = MSCB_FORMAT_ERROR continue if response[0] == MCMD_ACK + 7: if last is not None and response[1] & 0x80: payload = response[3:-1] else: payload = response[2:-1] else: payload = response[1:-1] if last is None and len(response) - 2 in (2, 4): payload = payload[::-1] return MSCB_SUCCESS, payload return status, b"" def _read_into(fd: int, adr: int, first: int, last: Optional[int], data: Any, size: Any, retries: int) -> int: capacity = _get_scalar(size, ctypes.c_int) if capacity > (1400 if last is not None else 256): return MSCB_INVAL_PARAM _copy_to_buffer(data, b"", capacity) status, payload = _read_value(fd, adr, first, last, retries) if status != MSCB_SUCCESS: return status if len(payload) > capacity: _set_scalar(size, ctypes.c_int, 0) return MSCB_NO_MEM _copy_to_buffer(data, payload, capacity) _set_scalar(size, ctypes.c_int, len(payload)) return MSCB_SUCCESS def mscb_read(fd: int, adr: int, index: int, data: Any, size: Any) -> int: return _read_into(fd, adr, index, None, data, size, _max_retry) def mscb_read_no_retries(fd: int, adr: int, index: int, data: Any, size: Any) -> int: return _read_into(fd, adr, index, None, data, size, 1) def mscb_read_range(fd: int, adr: int, index1: int, index2: int, data: Any, size: Any) -> int: return _read_into(fd, adr, index1, index2, data, size, _max_retry) def _memory_write(fd: int, node_adr: int, sub_adr: int, mem_adr: int, payload: bytes) -> int: if len(payload) > 1400: return MSCB_INVAL_PARAM count = len(payload) + 5 body = bytes((MCMD_WRITE_MEM, 0x80 | (count >> 8), count & 0xFF, _as_int(sub_adr) & 0xFF)) + struct.pack("!I", _as_int(mem_adr) & 0xFFFFFFFF) + payload data_crc = crc8(payload) status, response = _request_node(fd, node_adr, body + bytes((crc8(body),)), 270, RS485_FLAG_ADR_CYCLE | RS485_FLAG_LONG_TO) if status != MSCB_SUCCESS: return status if response == b"\xff": return MSCB_TIMEOUT return MSCB_SUCCESS if response == bytes((MCMD_ACK, data_crc)) else MSCB_CRC_ERROR def _memory_read(fd: int, node_adr: int, sub_adr: int, mem_adr: int, size: int) -> tuple[int, bytes]: if not 0 <= size <= 1400: return MSCB_INVAL_PARAM, b"" body = bytes((MCMD_READ_MEM, 0x07, size >> 8, size & 0xFF, _as_int(sub_adr) & 0xFF)) + struct.pack("!I", _as_int(mem_adr) & 0xFFFFFFFF) status, response = _request_node(fd, node_adr, body + bytes((crc8(body),)), size + 4, RS485_FLAG_ADR_CYCLE | RS485_FLAG_LONG_TO) if status != MSCB_SUCCESS: return status, b"" if response == b"\xff": return MSCB_TIMEOUT, b"" if len(response) != size + 4 or crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR, b"" return MSCB_SUCCESS, response[3:-1] def mscb_write_mem(fd: int, node_adr: int, sub_adr: int, mem_adr: int, buffer: Any, size: int) -> int: count = _as_int(size) if not 0 <= count <= 256: return MSCB_INVAL_PARAM return _memory_write(fd, node_adr, sub_adr, mem_adr, _as_bytes(buffer, count)) def mscb_read_mem(fd: int, node_adr: int, sub_adr: int, mem_adr: int, buffer: Any, size: int) -> int: count = _as_int(size) if not 0 <= count <= 256: return MSCB_INVAL_PARAM _copy_to_buffer(buffer, b"", count) status, payload = _memory_read(fd, node_adr, sub_adr, mem_adr, count) if status == MSCB_SUCCESS: _copy_to_buffer(buffer, payload, count) return status def mscb_user(fd: int, adr: int, param: Any, size: int, result: Any, rsize: Any) -> int: count = _as_int(size) capacity = _get_scalar(rsize, ctypes.c_int) if rsize is not None else 0 if not 0 <= count <= 4: return MSCB_FORMAT_ERROR body = bytes((MCMD_USER + count,)) + _as_bytes(param, count) status, response = _request_node(fd, adr, body + bytes((crc8(body),)), 80) if status != MSCB_SUCCESS: return status if len(response) < 2: return MSCB_TIMEOUT if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR payload = response[1:-1] _copy_to_buffer(result, payload, capacity) if rsize is not None: _set_scalar(rsize, ctypes.c_int, min(len(payload), capacity)) return MSCB_SUCCESS def mscb_echo(fd: int, adr: int, d1: int, d2: Any) -> int: _set_scalar(d2, ctypes.c_ubyte, 0xFF) body = bytes((MCMD_ECHO, _as_int(d1) & 0xFF)) status, response = _request_node(fd, adr, body + bytes((crc8(body),)), 64) if status != MSCB_SUCCESS: return status if response == b"\xff": return MSCB_TIMEOUT_BUS if len(response) < 3: return MSCB_TIMEOUT if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR _set_scalar(d2, ctypes.c_ubyte, response[1]) return MSCB_SUCCESS def mscb_link(fd: int, adr: int, index: int, data: Any, size: int) -> int: """Synchronize a caller-owned value with a node at most twice per second.""" key = (_as_int(fd), _as_int(adr), _as_int(index)) entry = _link_cache.get(key) if entry is None: info = MSCB_INFO_VAR() status = mscb_info_variable(fd, adr, index, ctypes.byref(info)) if status != MSCB_SUCCESS: return status status, payload = _read_value(fd, adr, index, None, _max_retry) if status != MSCB_SUCCESS: return status width = int(info.width) cached = payload[:width].ljust(width, b"\0") _link_cache[key] = (cached, width, time.monotonic()) _copy_to_buffer(data, cached, _as_int(size)) return MSCB_SUCCESS cached, width, last_read = entry supplied = _as_bytes(data, width) if supplied != cached: status = mscb_write(fd, adr, index, data, width) if status == MSCB_SUCCESS: _link_cache[key] = (supplied, width, last_read) return status if time.monotonic() - last_read >= 0.5: status, payload = _read_value(fd, adr, index, None, _max_retry) if status != MSCB_SUCCESS: return status cached = payload[:width].ljust(width, b"\0") _link_cache[key] = (cached, width, time.monotonic()) _copy_to_buffer(data, cached, _as_int(size)) return MSCB_SUCCESS def mscb_subm_reset(fd: int) -> int: if _connection(fd) is None: return MSCB_INVAL_PARAM _exchange(fd, bytes((MCMD_INIT,)), RS485_FLAG_CMD | RS485_FLAG_NO_ACK) time.sleep(5) status, response = _exchange(fd, bytes((MCMD_ECHO,)), RS485_FLAG_CMD, 10) if status != MSCB_SUCCESS or len(response) < 2: return MSCB_SUBM_ERROR _exchange(fd, b"\0" * 10, RS485_FLAG_BIT9 | RS485_FLAG_NO_ACK) return MSCB_SUCCESS def mscb_subm_info(fd: int) -> int: conn = _connection(fd) if conn is None: return MSCB_INVAL_PARAM status, response = _exchange(fd, bytes((MCMD_ECHO,)), RS485_FLAG_CMD | RS485_FLAG_NO_RETRY, 10) if status != MSCB_SUCCESS: return status if len(response) >= 4: print(f"Submaster : {conn.device}") print(f"Address : {socket.gethostbyname(conn.host)}") print(f"Protocol version : {response[1]}") print(f"Revision : 0x{int.from_bytes(response[2:4], 'big'):04X}") if len(response) == 8: uptime = int.from_bytes(response[4:8], "big") print(f"Uptime : {uptime // 86400}d {(uptime % 86400) // 3600:02d}h " f"{(uptime % 3600) // 60:02d}m {uptime % 60:02d}s") return MSCB_SUCCESS def mscb_set_time(fd: int, addr: int, gaddr: int, broadcast: int) -> int: now = time.localtime() fields = (now.tm_mday, now.tm_mon, now.tm_year % 100, now.tm_hour, now.tm_min, now.tm_sec) bcd = bytes(((value // 10) * 0x10 + value % 10) for value in fields) return _send_selected(fd, addr, gaddr, broadcast, bytes((MCMD_SET_TIME,)) + bcd) def mscb_clear_log(fd: int, adr: int) -> int: body = bytes((MCMD_LOG, 1)) for _ in range(_max_retry): status, response = _request_node(fd, adr, body + bytes((crc8(body),))) if status == MSCB_SUCCESS and len(response) == 2 and response[0] == MCMD_ACK: return MSCB_SUCCESS if crc8(response[:-1]) == response[-1] else MSCB_CRC_ERROR return MSCB_TIMEOUT def mscb_read_log(fd: int, adr: int, data_buffer: Any, bufsize: int) -> int: capacity = _as_int(bufsize) body = bytes((MCMD_LOG, 0)) response = b"" for _ in range(_max_retry): status, response = _request_node(fd, adr, body + bytes((crc8(body),)), capacity) if status == MSCB_SUCCESS and response and response[0] == MCMD_ACK + 7: break if len(response) < 2: return MSCB_TIMEOUT if crc8(response[:-1]) != response[-1]: return MSCB_CRC_ERROR _copy_to_buffer(data_buffer, response, capacity) return MSCB_SUCCESS def _filename(value: Any) -> Path: return Path(_decode_text(value)) def _parse_firmware(path: Path) -> tuple[int, bytes, bytes, bytes, str]: try: raw = path.read_bytes() except OSError: return MSCB_NOT_FOUND, b"", b"", b"", "" suffix = path.suffix.lower() if suffix == ".hex": image = bytearray(b"\xff" * 0x10000) image_size = 0 try: for line in raw.splitlines(): if not line.startswith(b":") or len(line) < 11: return MSCB_FORMAT_ERROR, b"", b"", b"", suffix count = int(line[1:3], 16) offset = int(line[3:7], 16) record_type = int(line[7:9], 16) if record_type == 0: data = bytes.fromhex(line[9:9 + count * 2].decode("ascii")) image[offset:offset + count] = data image_size += count elif record_type == 1: break return MSCB_SUCCESS, bytes(image[:image_size]), b"", b"", suffix except (ValueError, UnicodeError): return MSCB_FORMAT_ERROR, b"", b"", b"", suffix if suffix == ".bit": try: pos = 0 initial = int.from_bytes(raw[pos:pos + 2], "big"); pos += 2 + initial marker_len = int.from_bytes(raw[pos:pos + 2], "big"); pos += 2 if marker_len != 1 or raw[pos:pos + 1] != b"a": raise ValueError pos += marker_len offsets = [] for marker in (b"a", b"b", b"c", b"d"): if marker != b"a": if raw[pos:pos + 1] != marker: raise ValueError pos += 1 length = int.from_bytes(raw[pos:pos + 2], "big"); pos += 2 offsets.append(pos) pos += length if raw[pos:pos + 1] != b"e": raise ValueError pos += 1 length = int.from_bytes(raw[pos:pos + 4], "big"); pos += 4 if len(raw) - pos != length: raise ValueError values = offsets + [length, pos, XIL_BITFILE_HEADER_ID] checksum = (~sum(values)) & 0xFFFFFFFF preheader = struct.pack("!8I", *(values + [checksum])) return MSCB_SUCCESS, raw[pos:], raw[:pos], preheader, suffix except (IndexError, ValueError): return MSCB_FORMAT_ERROR, b"", b"", b"", suffix if suffix == ".srec": values = [16, len(raw), 17] preheader = struct.pack("!4I", *(values + [(~sum(values)) & 0xFFFFFFFF])) return MSCB_SUCCESS, raw, b"", preheader, suffix if suffix == ".bin": return MSCB_SUCCESS, raw, b"", b"", suffix return MSCB_FORMAT_ERROR, b"", b"", b"", suffix def _legacy_mode(fd: int, adr: int) -> int: status, response = _exchange(fd, bytes((MCMD_FREEZE, 1)), RS485_FLAG_CMD, 64) if status != MSCB_SUCCESS or len(response) != 1: return MSCB_TIMEOUT body = bytes((MCMD_UPGRADE,)) status, response = _request_node(fd, adr, body + bytes((crc8(body),)), 64, RS485_FLAG_LONG_TO | RS485_FLAG_ADR_CYCLE) if status != MSCB_SUCCESS or len(response) != 3: _exchange(fd, bytes((MCMD_FREEZE, 0)), RS485_FLAG_CMD, 64) return MSCB_TIMEOUT if response[1] == 2: _exchange(fd, bytes((MCMD_FREEZE, 0)), RS485_FLAG_CMD, 64) return MSCB_SUBADDR if response[1] == 3: _exchange(fd, bytes((MCMD_FREEZE, 0)), RS485_FLAG_CMD, 64) return MSCB_NOTREADY time.sleep(0.5) status, response = _exchange(fd, bytes((UCMD_ECHO,)), RS485_FLAG_LONG_TO, 64) if status != MSCB_SUCCESS or len(response) != 2: _exchange(fd, bytes((UCMD_RETURN,)), RS485_FLAG_NO_ACK) _exchange(fd, bytes((MCMD_FREEZE, 0)), RS485_FLAG_CMD, 64) return MSCB_TIMEOUT return MSCB_SUCCESS def _leave_legacy_mode(fd: int, reboot: bool) -> None: command = UCMD_REBOOT if reboot else UCMD_RETURN _exchange(fd, bytes((command,)), RS485_FLAG_NO_ACK) _exchange(fd, bytes((MCMD_FREEZE, 0)), RS485_FLAG_CMD, 64) mscb_clear_info_cache() def _legacy_upload(fd: int, adr: int, image: bytes) -> int: image = image.ljust(0x10000, b"\xff")[:0x10000] pages = [page for page in range(128) if any(octet != 0xFF for octet in image[page * 512:(page + 1) * 512])] status = _legacy_mode(fd, adr) if status != MSCB_SUCCESS: return status protected_page = 128 completed = False try: for page in pages: for _ in range(_max_retry): status, response = _exchange(fd, bytes((UCMD_ERASE, page)), RS485_FLAG_LONG_TO, 64) if status == MSCB_SUCCESS and len(response) == 2: if response[1] == 0xFF: protected_page = page break if response[0] == MCMD_ACK: break else: return MSCB_TIMEOUT if protected_page != 128: break for page in pages: if page >= protected_page: break page_data = image[page * 512:(page + 1) * 512] for _ in range(_max_retry): for subpage in range(16): chunk = page_data[subpage * 32:(subpage + 1) * 32] for _ in range(_max_retry): status, response = _exchange( fd, bytes((UCMD_PROGRAM, page, subpage)) + chunk, RS485_FLAG_LONG_TO, 64, ) if status == MSCB_SUCCESS and len(response) == 2 and response[0] == MCMD_ACK: break else: return MSCB_TIMEOUT checksum = sum(page_data) & 0xFF status, response = _exchange(fd, bytes((UCMD_VERIFY, page)), RS485_FLAG_LONG_TO, 64) if status == MSCB_SUCCESS and len(response) == 2 and response[1] == checksum: break else: return MSCB_CRC_ERROR completed = True return MSCB_SUCCESS finally: _leave_legacy_mode(fd, reboot=completed) def mscb_legacy_verify(fd: int, adr: int, buffer: Any) -> int: image = _as_bytes(buffer, 0x10000) if mscb_ping(fd, adr, 0, 1) != MSCB_SUCCESS: return MSCB_TIMEOUT status = _legacy_mode(fd, adr) if status != MSCB_SUCCESS: return status try: for page in range(128): page_data = image[page * 512:(page + 1) * 512] if not any(octet != 0xFF for octet in page_data): continue for subpage in range(16): expected = page_data[subpage * 32:(subpage + 1) * 32] for _ in range(5): status, response = _exchange(fd, bytes((UCMD_READ, page, subpage)), RS485_FLAG_LONG_TO, 64) if status == MSCB_SUCCESS and len(response) == 35: break else: return MSCB_TIMEOUT if response[2:34] != expected: return MSCB_CRC_ERROR return MSCB_SUCCESS finally: _leave_legacy_mode(fd, reboot=False) def mscb_upload(fd: int, node_adr: int, sub_adr: int, filename: Any, flags: int) -> int: status, image, header, preheader, suffix = _parse_firmware(_filename(filename)) if status != MSCB_SUCCESS: return status subaddressed = bool(_as_int(flags) & MSCB_UPLOAD_SUBADDR) if subaddressed: info = SimpleNamespace(buf_size=256, bootBank=0) else: status = mscb_ping(fd, node_adr, 0, 1) if status != MSCB_SUCCESS: return status status, info = _read_node_info(fd, node_adr) if status != MSCB_SUCCESS or info is None: return status if not info.buf_size: return _legacy_upload(fd, node_adr, image) if suffix == ".bit": status = _memory_write(fd, node_adr, sub_adr, XIL_FW_HEADER_ADDR | MSCB_BASE_FLASH, preheader + header) base = XIL_FW_ADDR | MSCB_BASE_FLASH elif suffix == ".srec": status = _memory_write(fd, node_adr, sub_adr, XIL_SW_HEADER_ADDR | MSCB_BASE_FLASH, preheader) base = XIL_SW_ADDR | MSCB_BASE_FLASH else: status = MSCB_SUCCESS base = MSCB_BASE_CODE if info.bootBank else MSCB_BASE_FLASH if status != MSCB_SUCCESS: return status chunk_size = min(max(int(info.buf_size), 1), 1400) for offset in range(0, len(image), chunk_size): payload = image[offset:offset + chunk_size] for _ in range(_max_retry): status = _memory_write(fd, node_adr, sub_adr, base + offset, payload) if status == MSCB_SUCCESS: break if status != MSCB_SUCCESS: return status mscb_ping(fd, node_adr, 0, 1) if not subaddressed and suffix not in (".bit", ".srec"): mscb_reboot(fd, node_adr, -1, 0) mscb_clear_info_cache() return MSCB_SUCCESS def mscb_verify(fd: int, node_adr: int, sub_adr: int, filename: Any, flags: int) -> int: status, image, header, preheader, suffix = _parse_firmware(_filename(filename)) if status != MSCB_SUCCESS: return status if _as_int(flags) & MSCB_UPLOAD_SUBADDR: info = SimpleNamespace(buf_size=256, bootBank=0) else: status, info = _read_node_info(fd, node_adr) if status != MSCB_SUCCESS or info is None: return status if not info.buf_size: legacy_image = (ctypes.c_ubyte * 0x10000).from_buffer_copy( image.ljust(0x10000, b"\xff")[:0x10000] ) return mscb_legacy_verify(fd, node_adr, legacy_image) if suffix == ".bit": base, expected_header = MSCB_BASE_FLASH | XIL_FW_HEADER_ADDR, preheader + header image_base = MSCB_BASE_FLASH | XIL_FW_ADDR elif suffix == ".srec": base, expected_header = MSCB_BASE_FLASH | XIL_SW_HEADER_ADDR, preheader image_base = MSCB_BASE_FLASH | XIL_SW_ADDR else: base, expected_header = 0, b"" image_base = MSCB_BASE_CODE if info.bootBank else MSCB_BASE_FLASH if expected_header: status, actual = _memory_read(fd, node_adr, sub_adr, base, len(expected_header)) if status != MSCB_SUCCESS: return status if actual != expected_header: return MSCB_CRC_ERROR chunk_size = min(max(int(info.buf_size), 1), 1400) for offset in range(0, len(image), chunk_size): expected = image[offset:offset + chunk_size] status, actual = _memory_read(fd, node_adr, sub_adr, image_base + offset, len(expected)) if status != MSCB_SUCCESS: return status if actual != expected: return MSCB_CRC_ERROR return MSCB_SUCCESS def mscb_download(fd: int, node_adr: int, sub_adr: int, filename: Any) -> int: path = _filename(filename) if path.exists(): return MSCB_NOT_FOUND status, info = _read_node_info(fd, node_adr) if status != MSCB_SUCCESS or info is None or not info.buf_size: return status if status != MSCB_SUCCESS else MSCB_NOTREADY suffix = path.suffix.lower() if suffix == ".bit": status, preheader = _memory_read(fd, node_adr, sub_adr, MSCB_BASE_FLASH | XIL_FW_HEADER_ADDR, 32) if status != MSCB_SUCCESS: return status values = struct.unpack("!8I", preheader) flash_size, header_size = values[4], values[5] status, stored = _memory_read(fd, node_adr, sub_adr, MSCB_BASE_FLASH | XIL_FW_HEADER_ADDR, 32 + header_size) if status != MSCB_SUCCESS: return status output = bytearray(stored[32:]) base = MSCB_BASE_FLASH | XIL_FW_ADDR elif suffix == ".srec": status, preheader = _memory_read(fd, node_adr, sub_adr, MSCB_BASE_FLASH | XIL_SW_HEADER_ADDR, 16) if status != MSCB_SUCCESS: return status _name, flash_size, _head, _checksum = struct.unpack("!4I", preheader) output = bytearray() base = MSCB_BASE_FLASH | XIL_SW_ADDR else: return MSCB_FORMAT_ERROR chunk_size = min(max(int(info.buf_size), 1), 1400) for offset in range(0, flash_size, chunk_size): count = min(chunk_size, flash_size - offset) status, payload = _memory_read(fd, node_adr, sub_adr, base + offset, count) if status != MSCB_SUCCESS: return status output.extend(payload) try: path.write_bytes(output) except OSError: return MSCB_NO_MEM return MSCB_SUCCESS def host2ip(hostname: Any, ip: Any, size: int) -> int: host = _decode_text(hostname).split(":", 1)[0] try: result = socket.gethostbyname(host) except OSError: _copy_to_buffer(ip, b"\0", _as_int(size)) return 0 _copy_to_buffer(ip, result.encode() + b"\0", _as_int(size)) return 1 def mscb_scan_udp() -> None: """Find Ethernet submasters named MSCB000 through MSCB999. This intentionally follows the C implementation: resolve each conventional hostname and send a direct submaster ECHO request to every name that exists. A subnet broadcast is not used because it is commonly blocked or ignored. """ sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sequence = 0 try: for number in range(1000): hostname = f"MSCB{number:03d}" print(f"Checking {hostname}...", end="\r", flush=True) try: address = socket.gethostbyname(hostname) except OSError: continue sequence = (sequence + 1) & 0xFFFF packet = _HEADER.pack( 1, sequence, RS485_FLAG_CMD | RS485_FLAG_NO_RETRY, MSCB_PROTOCOL_VERSION, ) + bytes((MCMD_ECHO,)) try: sock.sendto(packet, (address, MSCB_NET_PORT)) except OSError: continue deadline = time.monotonic() + 0.3 payload = b"" while time.monotonic() < deadline: sock.settimeout(max(0.0, deadline - time.monotonic())) try: response, peer = sock.recvfrom(1500) except (OSError, socket.timeout): break if peer[0] != address or len(response) < _HEADER.size: continue size, reply_sequence, _flags, version = _HEADER.unpack_from(response) if version != MSCB_PROTOCOL_VERSION or reply_sequence != sequence: continue if size + _HEADER.size != len(response): continue payload = response[_HEADER.size:] break if len(payload) >= 4: revision = int.from_bytes(payload[2:4], "big") message = f"Found {hostname}, PV {payload[1]}, Rev. 0x{revision:04X}" if len(payload) >= 8: uptime = int.from_bytes(payload[4:8], "big") message += ( f", UT {uptime // 86400}d " f"{(uptime % 86400) // 3600:02d}h " f"{(uptime % 3600) // 60:02d}m " f"{uptime % 60:02d}s" ) print(message) except KeyboardInterrupt: pass finally: sock.close() print(" " * 40) class MSCB: """Small Pythonic context-manager facade over the compatible functions.""" def __init__(self, device: str, password: str = "", debug: int = 0): self.device, self.password, self.debug = device, password, debug self.fd = -1 def connect(self) -> "MSCB": self.fd = mscb_init(self.device, len(self.device), self.password, self.debug) if self.fd < 0: raise ConnectionError(f"mscb_init failed with status {self.fd}") return self def close(self) -> None: if self.fd > 0: mscb_exit(self.fd) self.fd = -1 def __enter__(self) -> "MSCB": return self.connect() def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool: self.close() return False def ping(self, address: int, retry: bool = False) -> bool: return mscb_ping(self.fd, address, 0, int(retry)) == MSCB_SUCCESS def read(self, address: int, index: int, size: int = 256) -> bytes: return_status, payload = _read_value(self.fd, address, index, None, _max_retry) if return_status != MSCB_SUCCESS: raise OSError(return_status, "MSCB read failed") if len(payload) > size: raise BufferError("MSCB value exceeds requested size") return payload def write(self, address: int, index: int, data: bytes) -> None: status = _write_value(self.fd, address, index, bytes(data), False, _max_retry) if status != MSCB_SUCCESS: raise OSError(status, "MSCB write failed") __all__ = [name for name in globals() if name.startswith(("mscb_", "MSCB", "EMSCB", "MCMD_"))]