#!/usr/bin/env python3 """Python port of the MSCB command-line tool using the pure-Python MSCB API.""" from __future__ import annotations import argparse import ctypes import mscb from ctypes import byref from ctypes import c_int from ctypes import c_ubyte from ctypes import c_uint from ctypes import c_ulonglong from ctypes import create_string_buffer import os import select import struct import sys import time import xml.etree.ElementTree as ET from pathlib import Path from typing import Optional try: import msvcrt except ImportError: # pragma: no cover msvcrt = None try: import termios import tty except ImportError: # pragma: no cover termios = None tty = None MSCB_BASE_RAM = 0x00000000 MSCB_UPLOAD_SUBADDR = 0x02 MCMD_ADDR_NODE16 = 0x0A MSCB_SUCCESS = 1 MSCB_CRC_ERROR = 2 MSCB_TIMEOUT = 3 MSCB_TIMEOUT_BUS = 4 MSCB_FORMAT_ERROR = 7 MSCB_SUBM_ERROR = 9 MSCB_ADDR_EXISTS = 10 MSCB_SUBADDR = 12 MSCB_NOTREADY = 13 MSCB_NO_VAR = 14 MSCB_NOT_FOUND = 16 EMSCB_WRONG_PASSWORD = -8 EMSCB_COMM_ERROR = -9 EMSCB_NOT_FOUND = -10 EMSCB_NO_WRITE_ACCESS = -11 EMSCB_PROTOCOL_VERSION = -12 EMSCB_LOCKED = -5 MSCBF_FLOAT = 1 << 0 MSCBF_SIGNED = 1 << 1 MSCBF_DATALESS = 1 << 2 MSCBF_HIDDEN = 1 << 3 UNIT_STRING = 56 UNIT_ASCII = 55 UNIT_BYTE = 52 PREFIX_TABLE = { -12: "pico", -9: "nano", -6: "micro", -3: "milli", 0: "", 3: "kilo", 6: "mega", 9: "giga", 12: "tera", } UNIT_TABLE = { 1: "meter", 2: "gram", 3: "second", 4: "minute", 5: "hour", 6: "ampere", 7: "kelvin", 8: "deg. celsius", 9: "deg. farenheit", 20: "hertz", 21: "pascal", 22: "bar", 23: "watt", 24: "volt", 25: "ohm", 26: "tesla", 27: "liter/sec", 28: "RPM", 29: "farad", 30: "joule", 31: "newton", 50: "boolean", 52: "byte", 53: "word", 54: "dword", 55: "ascii", 56: "string", 57: "baud", 90: "percent", 91: "PPM", 92: "counts", 93: "factor", 94: "volt/volt", } MSCB_INFO = mscb.MSCB_INFO MSCB_INFO_VAR = mscb.MSCB_INFO_VAR def decode_cstr(raw: bytes) -> str: return raw.split(b"\0", 1)[0].decode("latin-1", errors="ignore") def parse_int(token: str) -> int: if token.lower().startswith("0x"): return int(token, 16) return int(token, 10) def sleep_ms(ms: int) -> None: time.sleep(ms / 1000.0) def stop_requested() -> bool: if msvcrt is None: if not sys.stdin.isatty(): return False rlist, _, _ = select.select([sys.stdin], [], [], 0) return bool(rlist) return bool(msvcrt.kbhit()) def clear_key_buffer() -> None: if msvcrt is None: if not sys.stdin.isatty(): return while True: rlist, _, _ = select.select([sys.stdin], [], [], 0) if not rlist: break os.read(sys.stdin.fileno(), 1024) return while msvcrt.kbhit(): msvcrt.getch() class _RawConsoleInput: def __init__(self) -> None: self.fd: Optional[int] = None self.old_attr = None def __enter__(self): if msvcrt is not None or termios is None or tty is None or not sys.stdin.isatty(): return self self.fd = sys.stdin.fileno() self.old_attr = termios.tcgetattr(self.fd) tty.setcbreak(self.fd) return self def __exit__(self, exc_type, exc, tb): if self.fd is not None and self.old_attr is not None and termios is not None: termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old_attr) self.fd = None self.old_attr = None def read_key() -> Optional[bytes]: if msvcrt is not None: if msvcrt.kbhit(): return msvcrt.getch() return None if not sys.stdin.isatty(): return None rlist, _, _ = select.select([sys.stdin], [], [], 0) if not rlist: return None return os.read(sys.stdin.fileno(), 1) def split_command(line: str) -> list[str]: tokens: list[str] = [] i = 0 n = len(line) while i < n: while i < n and line[i].isspace(): i += 1 if i >= n: break quote = None if line[i] in ('"', "'", "`"): quote = line[i] i += 1 buf = [] while i < n: ch = line[i] if quote is not None: if ch == quote: i += 1 break else: if ch.isspace(): break buf.append(ch) i += 1 if quote is None: while i < n and not line[i].isspace(): buf.append(line[i]) i += 1 tokens.append("".join(buf)) return tokens def print_help() -> None: print("Available commands:\n") print("addr Address individual node") print("baddr Address all nodes (broadcast)") print("baud Set baud rate of MSCB bus") print("debug 0/1/2 Turn debuggin off/on, 2=show on screen") print("echo [fc] Perform echo test [fast,continuous]") print("flash Flash parameters into EEPROM") print("gaddr Address group of nodes") print("info Retrive node info") print("load Load node variables") print("ping [r] Ping node and set address [repeat mode]") print("read [r ] [a] Read node variable [repeat mode] [all variables]") print("read - [r ] [a] Read range of node variables [repeat] [all]") print("reboot Reboot addressed node") print("sa Set node address of addressed node") print("save [first last] Save current node variables [save range]") print("scan [r] [a] [q] Scan bus for nodes [repeat mode] [all] [quick]") print("sg Set group address of addressed node(s)") print("sn Set node name (up to 16 characters)") print("sr Reset current submaster") print("submaster Show info for current submaster") print("sync Synchronize local time with node(s)") print("terminal Enter teminal mode for SCS-210") print("upload [slot] [debug] Upload new firmware to node in [slot] [with debug info]") print("download [slot] Download firmware from node in [slot] to file") print("memrd Read from memory in ") print("memwr Write to memory in ") print("mup Upload new firmware to nodes a1-a2") print("mwr Write a variable to nodes from to ") print("verify [adr] Compare current firmware in [adr] with file") print("version Display version number") print("write [r ] Write node variable") print("write - Write range of node variables") print("log [c] Read log / clear Log [c]") print() def format_channel(index: int, info: MSCB_INFO_VAR, payload: bytes, verbose: bool) -> str: name = decode_cstr(bytes(info.name)) if verbose: line = f"{index:3d}: {name:<17s} " else: line = f"{name}: " width = int(info.width) unit = int(info.unit) flags = int(info.flags) if unit == UNIT_STRING: data = payload[:width].split(b"\0", 1)[0] text = data.decode("latin-1", errors="backslashreplace") escaped = ( text.replace("\x01", "\\001") .replace("\x02", "\\002") .replace("\t", "\\t") .replace("\n", "\\n") .replace("\r", "\\r") ) if verbose: line += f"STR{width:02d} \"{escaped}\"" else: line += f"\"{escaped}\"" else: padded = payload[: max(width, 1)] + b"\0" * max(0, 8 - width) if width == 1: val_u = padded[0] bits = "".join("1" if (val_u & (0x80 >> i)) else "0" for i in range(8)) if flags & MSCBF_SIGNED: val_s = struct.unpack("b", bytes([val_u]))[0] line += (f" 8bit S {val_s:15d} (0x{val_u:02X}/{bits})" if verbose else f"{val_s:15d} (0x{val_u:02X}/{bits})") else: line += (f" 8bit U {val_u:15d} (0x{val_u:02X}/{bits})" if verbose else f"{val_u:15d} (0x{val_u:02X}/{bits})") elif width == 2: val_u = int.from_bytes(padded[:2], "little", signed=False) if flags & MSCBF_SIGNED: val_s = int.from_bytes(padded[:2], "little", signed=True) line += (f"16bit S {val_s:15d} (0x{val_u:04X})" if verbose else f"{val_s:15d} (0x{val_u:04X})") else: line += (f"16bit U {val_u:15d} (0x{val_u:04X})" if verbose else f"{val_u:15d} (0x{val_u:04X})") elif width == 3: val_u = int.from_bytes(padded[:3], "little", signed=False) val_s = val_u if val_u < (1 << 23) else val_u - (1 << 24) if flags & MSCBF_SIGNED: line += (f"24bit S {val_s:15d} (0x{val_u:06X})" if verbose else f"{val_s:15d} (0x{val_u:06X})") else: line += (f"24bit U {val_u:15d} (0x{val_u:06X})" if verbose else f"{val_u:15d} (0x{val_u:06X})") elif width == 4: val_u = int.from_bytes(padded[:4], "little", signed=False) if flags & MSCBF_FLOAT: val_f = struct.unpack(" None: if initial_cmd: if initial_cmd.startswith("@"): path = Path(initial_cmd[1:]) if not path.exists(): print(f"Command file {path} not found.") return with path.open("r", encoding="utf-8", errors="ignore") as fp: for raw in fp: line = raw.strip("\r\n") if not line: continue if self._execute_line(line): return else: self._execute_line(initial_cmd) return while True: prompt = self._prompt() try: line = input(prompt) except EOFError: print() break except KeyboardInterrupt: print() continue if self._execute_line(line): break def _prompt(self) -> str: if self.current_addr >= 0: return f"node{self.current_addr}(0x{self.current_addr:X})> " if self.current_group >= 0: return f"group{self.current_group}> " if self.broadcast: return "all> " return "> " def _execute_line(self, line: str) -> bool: params = split_command(line) if not params: return False cmd = self._resolve_command(params[0]) p = params + [""] * 10 if cmd == "help": print_help() return False if cmd == "version": lib = create_string_buffer(32) prot = create_string_buffer(32) mscb.mscb_get_version(lib, prot) print(f"MSCB library version : {decode_cstr(lib.raw)}") print(f"MSCB protocol version : {decode_cstr(prot.raw)}") return False if cmd == "debug": level = int(p[1]) if p[1] else 0 mscb.mscb_debug(level) print("Debugging level changed successfully.") return False if cmd == "scan": self._cmd_scan(p) return False if cmd == "info": self._cmd_info() return False if cmd == "ping": self._cmd_ping(p) return False if cmd == "addr": if not p[1]: print("Please specify node address") else: addr = parse_int(p[1]) mscb.mscb_addr(self.fd, MCMD_ADDR_NODE16, addr, 0, 10) self.current_addr = addr self.current_group = -1 self.broadcast = False return False if cmd == "baddr": self.current_addr = -1 self.current_group = -1 self.broadcast = True return False if cmd == "gaddr": if not p[1]: print("Please specify group address") else: addr = parse_int(p[1]) try: ref = input(f"Enter address of first node in group {addr}: ") except EOFError: ref = "0" self.reference_addr = int(ref or "0") self.current_addr = -1 self.current_group = addr self.broadcast = False return False if cmd == "sa": self._cmd_set_addr(p) return False if cmd == "sg": self._cmd_set_group(p) return False if cmd == "sn": self._cmd_set_name(params) return False if cmd == "baud": self._cmd_baud() return False if cmd == "write": self._cmd_write(p, line) return False if cmd == "mwrite": self._cmd_mwrite(p) return False if cmd == "read": self._cmd_read(p, line) return False if cmd == "save": self._cmd_save(p) return False if cmd == "load": self._cmd_load(p) return False if cmd == "terminal": self._cmd_terminal() return False if cmd == "flash": self._require_target() st = mscb.mscb_flash(self.fd, self.current_addr, self.current_group, int(self.broadcast)) if st != MSCB_SUCCESS: print(f"Error: {st}") return False if cmd == "upload": self._cmd_upload(p) return False if cmd == "download": self._cmd_download(p) return False if cmd == "mupload": self._cmd_mupload(p) return False if cmd == "verify": self._cmd_verify(p) return False if cmd == "memwr": self._cmd_memwr(p) return False if cmd == "memrd": self._cmd_memrd(p) return False if cmd == "reboot": self._require_target() st = mscb.mscb_reboot(self.fd, self.current_addr, self.current_group, int(self.broadcast)) if st != MSCB_SUCCESS: print(f"Error: {st}") return False if cmd == "sr": mscb.mscb_subm_reset(self.fd) return False if cmd == "submaster": mscb.mscb_subm_info(self.fd) return False if cmd == "sync": self._cmd_sync() return False if cmd == "echo": self._cmd_echo(p) return False if cmd == "user": self._cmd_user(p) return False if cmd == "log": self._cmd_log(p) return False if cmd in ("exit", "quit"): return True print(f"Unknown command {p[0]} {p[1]} {p[2]}") return False @staticmethod def _prefix_match(user_cmd: str, full_cmd: str) -> bool: uc = user_cmd.strip().lower() if not uc: return False return full_cmd.lower().startswith(uc) def _resolve_command(self, user_cmd: str) -> str: uc = user_cmd.strip().lower() if uc == "?" or uc.startswith("he"): return "help" # Keep original C command-check order so ambiguous prefixes resolve similarly. ordered = [ "version", "debug", "scan", "info", "ping", "addr", "baddr", "gaddr", "sa", "sg", "sn", "sm", "baud", "write", "mwrite", "read", "save", "load", "terminal", "flash", "upload", "download", "mupload", "verify", "memwr", "memrd", "reboot", "sr", "submaster", "sync", "echo", "user", "log", "t1", "t2", "c1", "exit", "quit", ] for full in ordered: if self._prefix_match(uc, full): return full return uc def _require_target(self) -> bool: if self.current_addr < 0 and self.current_group < 0 and not self.broadcast: print("You must first address node (s)") return False return True def _cmd_scan(self, p: list[str]) -> None: try: while True: n_found = 0 quick = p[1].startswith("q") or p[2].startswith("q") for i in range(-1, 0x10000): if i != -1 and not (p[1].startswith("a") or p[2].startswith("a")) and i > 0 and not self.ping_addr[i]: continue if i == -1: print("Test address 65535 (0xFFFF)\r", end="", flush=True) status = mscb.mscb_ping(self.fd, 0xFFFF, 0, 0) sleep_ms(100) else: print(f"Test address {i:05d} (0x{i:04X})\r", end="", flush=True) status = mscb.mscb_ping(self.fd, i, int(quick), 0) if status == MSCB_SUCCESS: n_found += 1 for j in range(i, min(i + 100, 0x10000)): if j >= 0: self.ping_addr[j] = 1 info = MSCB_INFO() st2 = mscb.mscb_info(self.fd, i & 0xFFFF, byref(info)) name = decode_cstr(bytes(info.node_name))[:16] if st2 == MSCB_SUCCESS: print( f"Found node \"{name}\", NA {i & 0xFFFF} (0x{i & 0xFFFF:04X}), " f"GA {info.group_address} (0x{info.group_address:04X}), Rev. 0x{info.revision:04X} " ) elif status == MSCB_SUBM_ERROR: print("Error: Submaster not responding") break if stop_requested(): break print(" ") if n_found == 0: print("No nodes found ") elif n_found == 1: print("One node found ") else: print(f"{n_found} nodes found ") if not (p[1].startswith("r") or p[2].startswith("r")): break if stop_requested(): break finally: clear_key_buffer() def _cmd_info(self) -> None: if self.current_addr < 0: print("You must first address an individual node") return info = MSCB_INFO() status = mscb.mscb_info(self.fd, self.current_addr, byref(info)) if status == MSCB_CRC_ERROR: print("CRC Error") return if status != MSCB_SUCCESS: print("No response from node") return name = decode_cstr(bytes(info.node_name))[:16] print(f"Node name : {name}") print(f"Node address : {info.node_address} (0x{info.node_address:X})") print(f"Group address : {info.group_address} (0x{info.group_address:X})") print(f"Protocol version : {info.protocol_version}") if info.pinExt_resets == 0: print(f"Revision : 0x{info.revision:04X}") else: print(f"Revision : V {info.revision >> 8}.{(info.revision >> 4) & 0xF}.{info.revision & 0xF}") rtc0 = int(info.rtc[0]) if rtc0 and rtc0 != 0xFF: rtc = [] for i in range(6): v = int(info.rtc[i]) rtc.append((v // 0x10) * 10 + (v % 0x10)) print(f"Real Time Clock : {rtc[0]:02d}-{rtc[1]:02d}-{rtc[2]:02d} {rtc[3]:02d}:{rtc[4]:02d}:{rtc[5]:02d}") uptime = c_uint(0) if mscb.mscb_uptime(self.fd, self.current_addr, byref(uptime)) == MSCB_SUCCESS: u = int(uptime.value) print(f"Uptime : {u // (3600 * 24)}d {(u % (3600 * 24)) // 3600:02d}h {(u % 3600) // 60:02d}m {u % 60:02d}s") if info.buf_size != 0: print(f"Buffer size : {info.buf_size}") if info.bootBank != 0: print(f"Pin/Ext WD Resets : {info.pinExt_resets}") print(f"SW-resets : {info.SW_resets}") print(f"Int. WD resets : {info.int_WD_resets}") print(f"Boot-Bank : {info.bootBank}") print(f"Silicon Revision : {chr(info.siliconRevison)}") print(f"System load : {info.systemLoad:4.2f}%") print(f"Peak system load : {info.peakSystemLoad:4.2f}%") print(f"PCB Temp : {info.pcbTemp:4.2f} C") print(f"Supply 1.8V : {info.Supply1V8:4.2f}V") print(f"Supply 3.3V : {info.Supply3V3:4.2f}V") print(f"Supply 5.0V : {info.Supply5V0:4.2f}V") print(f"Supply 24.0V : {info.Supply24V0:4.2f}V") print(f"Supply 5V Ext. : {info.Supply5V0Ext:4.2f}V") print(f"Supply 24V Ext. : {info.Supply24V0Ext:4.2f}V") print(f"Supply current : {info.Supply24V0Current:4.3f}A") print(f"Battery voltage : {info.Vbat:4.2f}V") def _cmd_ping(self, p: list[str]) -> None: if not p[1]: print("Please specify node address") return addr = parse_int(p[1]) try: while True: status = mscb.mscb_ping(self.fd, addr, 0, 1) if status != MSCB_SUCCESS: if status == MSCB_SUBM_ERROR: print("Error: Submaster not responding") else: print(f"Node {addr} does not respond") self.current_addr = -1 self.current_group = -1 self.broadcast = False else: print(f"Node {addr} addressed") self.current_addr = addr self.current_group = -1 self.broadcast = False if not p[2].startswith("r"): break if stop_requested(): break sleep_ms(1000) finally: clear_key_buffer() def _cmd_set_addr(self, p: list[str]) -> None: if self.current_addr < 0 and self.current_group < 0 and not self.broadcast: print("You must first address node(s)") return if not p[1]: print("Please specify node address") return addr = parse_int(p[1]) st = mscb.mscb_set_node_addr(self.fd, self.current_addr, self.current_group, int(self.broadcast), addr) if self.current_addr >= 0: if st == MSCB_ADDR_EXISTS: print(f"Error: Address {addr} exists already on this network") else: self.current_addr = addr def _cmd_set_group(self, p: list[str]) -> None: if self.current_addr < 0 and self.current_group < 0 and not self.broadcast: print("You must first address node(s)") return if not p[1]: print("Please specify group address") return addr = parse_int(p[1]) mscb.mscb_set_group_addr(self.fd, self.current_addr, self.current_group, int(self.broadcast), addr) def _cmd_set_name(self, params: list[str]) -> None: if self.current_addr < 0: print("You must first address an individual node") return if len(params) < 2: print("Please specify node name") return name = " ".join(params[1:]).strip() if len(name) > 15: print("Maximum length is 15 characters, please choose shorter name") return mscb.mscb_set_name(self.fd, self.current_addr, name.encode("latin-1", errors="ignore")) def _cmd_baud(self) -> None: print("Possible baud rates:\n") print(" 1: 2400") print(" 2: 4800") print(" 3: 9600") print(" 4: 19200") print(" 5: 28800") print(" 6: 38400") print(" 7: 57600") print(" 8: 115200 <= default") print(" 9: 172800") print(" 10: 345600") rate = input("\nSelect rate: ") mscb.mscb_set_baud(self.fd, int(rate or "0")) def _addr_for_write(self) -> int: if self.current_addr < 0: return self.reference_addr return self.current_addr def _cmd_write(self, p: list[str], raw_line: str) -> None: if self.current_addr < 0 and self.current_group < 0: print("You must first address a node or a group") return if not p[1]: print("Please specify channel number") return idx1 = int(p[1].split("-")[0]) idx2 = int(p[1].split("-")[1]) if "-" in p[1] else idx1 idx1 = min(idx1, 255) idx2 = min(idx2, 255) for index in range(idx1, idx2 + 1): info = MSCB_INFO_VAR() status = mscb.mscb_info_variable(self.fd, self._addr_for_write(), index, byref(info)) if status == MSCB_NO_VAR: print(f"Node has no variable #{index}") break if status == MSCB_CRC_ERROR: print("CRC Error") return if status != MSCB_SUCCESS: print("Timeout or invalid channel number") return if info.unit in (UNIT_STRING, UNIT_ASCII): val = p[2].encode("latin-1", errors="ignore") + b"\0" buf = create_string_buffer(val) while True: if self.current_addr >= 0: status = mscb.mscb_write(self.fd, self.current_addr, index, buf, len(val)) else: status = mscb.mscb_write_group(self.fd, self.current_group, index, buf, len(val)) if not p[3]: break if stop_requested(): break sleep_ms(100 if info.unit == UNIT_ASCII else 1000) clear_key_buffer() else: if not p[2]: print("Please specify data") break width = int(info.width) flags = int(info.flags) write_size = max(width, 1) if flags & MSCBF_FLOAT: # Match native msc behavior: pass IEEE754 bits through an integer payload. fbits = struct.unpack("= 0: status = mscb.mscb_write( self.fd, self.current_addr, index, byref(payload_q), write_size ) else: status = mscb.mscb_write_group( self.fd, self.current_group, index, byref(payload_q), write_size ) if p[3].startswith("r") and p[4] and int(p[4]) > 0: sleep_ms(int(p[4])) print(f"\r{wheel[wi]} ", end="", flush=True) wi = (wi + 1) % 4 else: break if stop_requested(): break if p[3].startswith("r"): print() clear_key_buffer() if status != MSCB_SUCCESS: print(f"Error: {status}") def _cmd_mwrite(self, p: list[str]) -> None: if not (p[1] and p[2] and p[3] and p[4]): print("Please specify variable index, value, and node range") return index = int(p[1]) first = int(p[3]) last = int(p[4]) info = MSCB_INFO_VAR() mscb.mscb_info_variable(self.fd, first, index, byref(info)) status = MSCB_SUCCESS for addr in range(first, last + 1): print(f"Node {addr}") if info.unit in (UNIT_STRING, UNIT_ASCII): val = p[2].encode("latin-1", errors="ignore") + b"\0" buf = create_string_buffer(val) status = mscb.mscb_write(self.fd, addr, index, buf, len(val)) else: width = max(int(info.width), 1) if int(info.flags) & MSCBF_FLOAT: fbits = struct.unpack(" None: if self.current_addr < 0: print("You must first address an individual node") return if not p[1] or p[1].startswith("a") or p[1].startswith("r"): first, last = 0, 255 elif "-" in p[1]: first, last = int(p[1].split("-")[0]), int(p[1].split("-")[1]) else: first = last = int(p[1]) first = min(first, 255) last = min(last, 255) read_all = p[1].startswith("a") or p[2].startswith("a") or p[3].startswith("a") repeat = p[1].startswith("r") or p[2].startswith("r") or p[3].startswith("r") wait = 0 if repeat: for i in range(1, 4): if p[i].startswith("r") and p[i + 1] and p[i + 1].isdigit(): wait = int(p[i + 1]) break info_arr = {} n_found = 0 for i in range(first, last + 1): iv = MSCB_INFO_VAR() status = mscb.mscb_info_variable(self.fd, self.current_addr, i, byref(iv)) if status == MSCB_NO_VAR: if first == last: print(f"Node has no variable #{first}") break if status == MSCB_CRC_ERROR: print("CRC Error") return if status != MSCB_SUCCESS: print("Timeout or invalid channel number") return info_arr[i] = iv n_found += 1 if n_found == 0: return try: while True: if first == last: i = first size = c_int(max(int(info_arr[i].width), 1)) buf = create_string_buffer(size.value) status = mscb.mscb_read(self.fd, self.current_addr, i, buf, byref(size)) if status == MSCB_SUCCESS: print(format_channel(i, info_arr[i], buf.raw[:size.value], verbose=False), end="") else: end = first + n_found - 1 size = c_int(1400) dbuf = create_string_buffer(size.value) status = mscb.mscb_read_range(self.fd, self.current_addr, first, end, dbuf, byref(size)) if status == MSCB_CRC_ERROR: print("CRC Error") return if status == MSCB_TIMEOUT: print("Timeout receivig acknowledge") return if status != MSCB_SUCCESS: print("Reading not successful!") return pos = 0 for i in range(first, end + 1): w = int(info_arr[i].width) payload = bytearray(dbuf.raw[pos: pos + w]) if w == 2: payload[0], payload[1] = payload[1], payload[0] elif w == 4: payload[0], payload[3] = payload[3], payload[0] payload[1], payload[2] = payload[2], payload[1] if (int(info_arr[i].flags) & MSCBF_HIDDEN) == 0 or read_all or first > 0: print(format_channel(i, info_arr[i], bytes(payload), verbose=True), end="") pos += w if not repeat: break if stop_requested(): break if first != last: print() sleep_ms(wait if wait else 10) if wait: print() finally: if first == last: print() clear_key_buffer() def _save_node_xml(self, root: ET.Element, addr: int) -> None: info = MSCB_INFO() status = mscb.mscb_info(self.fd, addr, byref(info)) if status == MSCB_CRC_ERROR: print("CRC Error") return if status != MSCB_SUCCESS: print("No response from node") return node = ET.SubElement(root, "Node") ET.SubElement(node, "Name").text = decode_cstr(bytes(info.node_name)) ET.SubElement(node, "NodeAddress").text = f"{addr} (0x{addr:X})" ET.SubElement(node, "GroupAddress").text = f"{info.group_address} (0x{info.group_address:X})" ET.SubElement(node, "ProtocolVersion").text = f"{info.protocol_version // 16}.{info.protocol_version % 16}" vars_el = ET.SubElement(node, "Variables") for i in range(int(info.n_variables)): iv = MSCB_INFO_VAR() mscb.mscb_info_variable(self.fd, addr, i, byref(iv)) size = c_int(256) buf = create_string_buffer(256) mscb.mscb_read(self.fd, addr, i, buf, byref(size)) var_el = ET.SubElement(vars_el, "Variable") ET.SubElement(var_el, "Index").text = str(i) ET.SubElement(var_el, "Name").text = decode_cstr(bytes(iv.name))[:8] ET.SubElement(var_el, "Width").text = f"{int(iv.width) * 8}bit" line = format_channel(i, iv, buf.raw[: size.value], verbose=True) flags = line[29:].split(" ", 1)[0].strip() ET.SubElement(var_el, "Flags").text = flags data_part = line[29:].strip() value_tok = data_part.split(" ", 1)[1] if " " in data_part else data_part ET.SubElement(var_el, "Value").text = value_tok.split(" ")[0] if "(" in line and ")" in line: hb = line.split("(", 1)[1].split(")", 1)[0] ET.SubElement(var_el, "HexBinValue").text = hb unit = "" for u in UNIT_TABLE.values(): if line.rstrip().endswith(u): unit = u break ET.SubElement(var_el, "Unit").text = unit def _cmd_save(self, p: list[str]) -> None: if self.current_addr < 0 and not p[1] and not p[2]: print("You must first address an individual node or specify a range") return if not p[1] or "." not in p[1]: out = input("Enter file name: ") else: out = p[1] if not out: return if not p[2]: first = last = self.current_addr elif p[3]: first = int(p[2]) last = int(p[3]) else: first = int(p[1]) last = int(p[2]) root = ET.Element("MSCBDump") saved = 0 for i in range(first, last + 1): st = mscb.mscb_ping(self.fd, i, 0, 1) if st == MSCB_SUCCESS: saved += 1 print(f"Save node {i} (0x{i:04X}) \r", end="", flush=True) self._save_node_xml(root, i) else: print(f"Test address {i} (0x{i:04X}) \r", end="", flush=True) print() if saved > 1: print(f"{saved} nodes saved ") tree = ET.ElementTree(root) ET.indent(tree, space=" ") with open(out, "wb") as fp: tree.write(fp, encoding="utf-8", xml_declaration=True) fp.write(b"\n") def _cmd_load(self, p: list[str]) -> None: filename = p[1] if p[1] else input("Enter file name: ") if not filename: return yn = input("Write parameters to flash? (y/[n]) ") flash = yn.lower().startswith("y") tree = ET.parse(filename) root = tree.getroot() if root.tag != "MSCBDump": print(f"Error loading \"{filename}\": No MSCBDump structure in file") return for idx, node in enumerate(root.findall("Node")): naddr = node.findtext("NodeAddress", default="0") addr = int(naddr.split()[0]) print(f"\nLoading node #{naddr}") info = MSCB_INFO() if mscb.mscb_info(self.fd, addr, byref(info)) != MSCB_SUCCESS: print(f"No response from node {addr}, skip node.") continue chn_name = {} for i in range(int(info.n_variables)): iv = MSCB_INFO_VAR() mscb.mscb_info_variable(self.fd, addr, i, byref(iv)) chn_name[decode_cstr(bytes(iv.name))] = i vars_root = node.find("Variables") if vars_root is None: print(f"Error loading \"{filename}\": Node #{idx} does not contain Variables") return for var in vars_root.findall("Variable"): name = var.findtext("Name", default="") value = var.findtext("Value", default="") if name not in chn_name: print(f"Variable \"{name}\" from file not in node, variable skipped") continue i = chn_name[name] iv = MSCB_INFO_VAR() mscb.mscb_info_variable(self.fd, addr, i, byref(iv)) if iv.unit == UNIT_STRING: payload = value.encode("latin-1", errors="ignore")[: int(iv.width)] buf = create_string_buffer(payload + b"\0") st = mscb.mscb_write(self.fd, addr, i, buf, int(iv.width)) if st != MSCB_SUCCESS: print(f"Error writing to node {addr}, variable {i}") else: if int(iv.flags) & MSCBF_FLOAT: payload = struct.pack(" None: if self.current_addr < 0: print("You must first address an individual node") return chn = -1 iv = MSCB_INFO_VAR() for i in range(255): st = mscb.mscb_info_variable(self.fd, self.current_addr, i, byref(iv)) if st == MSCB_NO_VAR: break if int(iv.flags) & MSCBF_DATALESS: chn = i break if chn < 0: print("Node does not support terminal mode.\n") return print("Exit with \n") try: with _RawConsoleInput(): while True: c = read_key() if c is not None: if c == b"\x1b": break if c in (b"\r", b"\n"): sys.stdout.write("\n") else: sys.stdout.write(c.decode("latin-1", errors="ignore")) sys.stdout.flush() b = c[0] mscb.mscb_write(self.fd, self.current_addr, chn, byref(c_ubyte(b)), 1) if b == 13: nl = c_ubyte(10) mscb.mscb_write(self.fd, self.current_addr, chn, byref(nl), 1) buf = create_string_buffer(256) size = c_int(256) mscb.mscb_read(self.fd, self.current_addr, chn, buf, byref(size)) if size.value > 0: sys.stdout.write(buf.raw[: size.value].decode("latin-1", errors="ignore")) sys.stdout.flush() sleep_ms(10) finally: print("\n") clear_key_buffer() def _cmd_upload(self, p: list[str]) -> None: is_slot_mode = p[1].isdigit() if is_slot_mode: filename = p[2] if p[2] else input("Enter name of file: ") if not filename: return flags = 0 if p[2].lower() == "debug" or p[3].lower() == "debug": flags = 1 st = mscb.mscb_upload(self.fd, self.current_addr, int(p[1]), filename.encode(), flags | MSCB_UPLOAD_SUBADDR) else: if self.current_addr < 0: print("You must first address an individual node") return filename = p[1] if p[1] else input("Enter name of file: ") if not filename: return st = mscb.mscb_upload(self.fd, self.current_addr, 0, filename.encode(), int(p[2].startswith("d"))) self._print_upload_error(st, filename, self.current_addr) def _cmd_download(self, p: list[str]) -> None: is_slot_mode = p[1].isdigit() if is_slot_mode: filename = p[2] if p[2] else input("Enter name of file: ") if not filename: return st = mscb.mscb_download(self.fd, self.current_addr, int(p[1]), filename.encode()) else: if self.current_addr < 0: print("You must first address an individual node") return filename = p[1] if p[1] else input("Enter name of file: ") if not filename: return st = mscb.mscb_download(self.fd, self.current_addr, 0, filename.encode()) if st == MSCB_NOT_FOUND: print(f"File \"{filename}\" exists already") elif st == MSCB_TIMEOUT: print(f"Node {self.current_addr} does not respond") elif st == MSCB_SUBADDR: print("Cannot read subaddress") def _cmd_mupload(self, p: list[str]) -> None: first = int(p[2] or "0") last = int(p[3] or "0") if last == 0: print("You must specify an address range") return filename = p[1] for i in range(first, last + 1): print(f"Node{i}: ", end="") st = mscb.mscb_upload(self.fd, i, 0, filename.encode(), 0) self._print_upload_error(st, filename, self.current_addr) if st in (MSCB_NOT_FOUND, MSCB_FORMAT_ERROR): break def _cmd_verify(self, p: list[str]) -> None: is_slot_mode = p[1].isdigit() if is_slot_mode: filename = p[2] if p[2] else input("Enter name of file: ") if not filename: return st = mscb.mscb_verify(self.fd, self.current_addr, int(p[1]), filename.encode(), MSCB_UPLOAD_SUBADDR) else: if self.current_addr < 0: print("You must first address an individual node") return filename = p[1] if p[1] else input("Enter name of file: ") if not filename: return st = mscb.mscb_verify(self.fd, self.current_addr, 0, filename.encode(), 0) self._print_upload_error(st, filename, self.current_addr) def _cmd_memwr(self, p: list[str]) -> None: if not self._require_target(): return slot = int(p[1] or "0") mem_addr = parse_int(p[2] or "0") | MSCB_BASE_RAM value = p[3] if value.lower().startswith("0x"): raw = value[2:] if len(raw) <= 2: size = 1 data = int(value, 16).to_bytes(1, "big") elif len(raw) <= 4: size = 2 data = int(value, 16).to_bytes(2, "big") else: size = 4 data = int(value, 16).to_bytes(4, "big") else: size = 4 data = int(value).to_bytes(4, "big", signed=False) buf = create_string_buffer(data, len(data)) mscb.mscb_write_mem(self.fd, self.current_addr, slot, mem_addr, buf, size) def _cmd_memrd(self, p: list[str]) -> None: if not self._require_target(): return slot = int(p[1] or "0") mem_addr = parse_int(p[2] or "0") | MSCB_BASE_RAM size = parse_int(p[3] or "0") buf = create_string_buffer(size) st = mscb.mscb_read_mem(self.fd, self.current_addr, slot, mem_addr, buf, size) if st == MSCB_SUCCESS: hexs = " ".join(f"{b:02X}" for b in buf.raw[:size]) print(f"0x{mem_addr:02X}: {hexs}") def _cmd_sync(self) -> None: if not self._require_target(): return st = mscb.mscb_set_time(self.fd, self.current_addr, self.current_group, int(self.broadcast)) if st != MSCB_SUCCESS: print(f"Error: {st}") else: now = time.localtime() print( f"Synchornized to {now.tm_mday:02d}-{now.tm_mon:02d}-{now.tm_year - 2000:02d} " f"{now.tm_hour:02d}:{now.tm_min:02d}:{now.tm_sec:02d}" ) def _cmd_echo(self, p: list[str]) -> None: if self.current_addr < 0: print("You must first address an individual node") return d1 = 0 i = 0 try: while not stop_requested(): d1 = (d1 + 1) % 256 d2 = c_ubyte(0) st = mscb.mscb_echo(self.fd, self.current_addr, d1, byref(d2)) if st == MSCB_TIMEOUT: print("Timeout in submaster communictation") elif st == MSCB_TIMEOUT_BUS: print("Timeout from RS485 bus") elif st == MSCB_CRC_ERROR: print("CRC Error on RS485 bus") elif st != MSCB_SUCCESS: print(f"Error: {st}") if d2.value != d1: print(f"{i}\nReceived: {d2.value:02X}, should be {d1:02X}, status = {st}") if not (p[1].startswith("c") or (len(p[1]) > 1 and p[1][1] == "c")): break i += 1 if i % 100 == 0: print(f"{i}\r", end="", flush=True) if not p[1].startswith("f") and not (len(p[1]) > 1 and p[1][1] == "c"): sleep_ms(10) finally: print(i) clear_key_buffer() def _cmd_user(self, p: list[str]) -> None: if not self._require_target(): return result = c_int(0) rsize = c_int(ctypes.sizeof(result)) if p[1]: param = c_int(int(p[1])) st = mscb.mscb_user(self.fd, self.current_addr, byref(param), 1, byref(result), byref(rsize)) else: st = mscb.mscb_user(self.fd, self.current_addr, None, 0, byref(result), byref(rsize)) if st != MSCB_SUCCESS: print(f"Error: {st}") def _cmd_log(self, p: list[str]) -> None: if self.broadcast: print("Log command not allowed in broadcast mode!") return if self.current_group > 0: print("Log command not allowed in group address mode!") return if self.current_addr < 0: print("You must first address a single node") return info = MSCB_INFO() st = mscb.mscb_info(self.fd, self.current_addr, byref(info)) if st == MSCB_CRC_ERROR: print("CRC Error") return if st != MSCB_SUCCESS: print("node does not respond.") return if info.bootBank == 0: print("Log command not supported!") return if p[1].startswith("c"): st = mscb.mscb_clear_log(self.fd, self.current_addr) else: dbuf = create_string_buffer(100 * 1024) while True: st = mscb.mscb_read_log(self.fd, self.current_addr, dbuf, ctypes.sizeof(dbuf)) if st == MSCB_SUCCESS: msg = decode_cstr(dbuf.raw[3:]) print(msg, end="") if not msg: break else: break if st != MSCB_SUCCESS: print(f"Error: {st}") def _print_upload_error(self, status: int, filename: str, addr: int) -> None: if status == MSCB_NOT_FOUND: print(f"File \"{filename}\" not found") elif status == MSCB_FORMAT_ERROR: print(f"Syntax error in file \"{filename}\"") elif status == MSCB_TIMEOUT: print(f"Node {addr} does not respond") elif status == MSCB_SUBADDR: print("Cannot program subaddress") elif status == MSCB_NOTREADY: print("Note just rebooted and not ready for upgrade") def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(add_help=False) parser.add_argument("-d", dest="device", default="") parser.add_argument("-p", dest="password", default="") parser.add_argument("-a", dest="addr", default=0, type=int) parser.add_argument("-c", dest="command", default="") parser.add_argument("-v", dest="verbose", action="store_true") parser.add_argument("-w", dest="write_log", action="store_true") parser.add_argument("-s", dest="scan", action="store_true") parser.add_argument("-h", "--help", action="store_true", dest="help") return parser.parse_args(argv) def usage() -> None: print("usage: msc.py [-d host[:port]] [-p password] [-a addr] [-c Command] [-c @CommandFile] [-v] [-s]\n") print(' -d Ethernet submaster hostname/IP, optionally followed by UDP port') print(" -p optional password for SUBM_260") print(" -s Scan for Ethernet submasters on local net") print(" -a Address node before executing command") print(" -c Execute command immediately") print(" -v Produce verbose debugging output into mscb_debug.log") print(" -w Log all write commands to mscb_write.log") print("\nFor a list of valid commands start msc.py interactively and type \"help\".") def main(argv: list[str]) -> int: args = parse_args(argv) if args.help: usage() return 0 if args.scan: mscb.mscb_scan_udp() return 0 if not args.device: print("Please specify MSCB submaster via -d flag") return 0 ip_buf = create_string_buffer(256) try: mscb.host2ip(args.device.encode(), ip_buf, ctypes.sizeof(ip_buf)) except OSError: ip_buf.value = b"" ip = decode_cstr(ip_buf.raw) debug = 2 if args.verbose else (1 if args.write_log else 0) fd = mscb.mscb_init(args.device.encode(), len(args.device), args.password.encode(), debug) if fd == EMSCB_WRONG_PASSWORD: pw = input(f"Enter password to access {args.device}: ").strip() fd = mscb.mscb_init(args.device.encode(), len(args.device), pw.encode(), debug) if fd < 0: if fd == EMSCB_COMM_ERROR: print(f"\nCannot communicate with MSCB submaster {args.device} ({ip})") print("Please disconnect and reconnect submaster\n") elif fd == EMSCB_PROTOCOL_VERSION: print(f"\nSubmaster {args.device} ({ip}) runs old protocol version") print("Please upgrade submaster software\n") elif fd == EMSCB_WRONG_PASSWORD: print("\nWrong password") elif fd == EMSCB_LOCKED: print("\nMSCB system is locked by other process") print("Please stop all running MSCB clients\n") else: print(f"Cannot connect to device \"{args.device}\" ({ip})") return 0 print(f"Connected to submaster {args.device} ({ip})") try: cli = MscCli(fd, args.device, ip, args.addr) cli.run(args.command) finally: mscb.mscb_exit(fd) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))