diff --git a/app_state.py b/app_state.py index 7178c39..8aa37c5 100644 --- a/app_state.py +++ b/app_state.py @@ -8,7 +8,7 @@ so later rebinding would only ever be seen by the module that did it. import configparser import os -VERSION = "0.0.0.072" +VERSION = "0.0.0.074" PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties") DEFAULT_GRID_SIZE = 10 DEFAULT_ADD_TIMESTAMP = True @@ -204,4 +204,17 @@ def mark_clean(): refresh_dirty_display() +# ── Live device connection state ──────────────────────────────── +live_mode = False +device_session = None # device_session.DeviceSession instance +device_host = "" +device_port = 1876 +connection_status = "offline" # offline | connecting | connected | error + + +def base_window_title_live(): + """Window title for live mode.""" + return f"Sedona Live Editor [v{VERSION}] — {device_host}:{device_port}" + + SEDONA_HOME, GRID_SIZE, ADD_TIMESTAMP, BOX_WIDTH, LEFT_PANE_WIDTH = load_preferences() diff --git a/connect_dialog.py b/connect_dialog.py new file mode 100644 index 0000000..8bb4957 --- /dev/null +++ b/connect_dialog.py @@ -0,0 +1,169 @@ +""" +Connection dialog for connecting to a live Sedona device. + +Modal Toplevel with host, port, username, password fields. +Connect runs in a background thread; progress and errors shown inline. +""" + +import configparser +import threading +import tkinter as tk +from tkinter import ttk + +import app_state + + +class ConnectDialog: + """Modal dialog for entering device connection parameters.""" + + def __init__(self, parent, on_connect_callback): + self._callback = on_connect_callback + self._thread = None + self._cancelled = False + + self._dlg = tk.Toplevel(parent) + self._dlg.title("Connect to Device") + self._dlg.resizable(False, False) + self._dlg.transient(parent) + self._dlg.grab_set() + + host, port, user = _load_last_connection() + + frame = ttk.Frame(self._dlg, padding=16) + frame.pack(fill=tk.BOTH, expand=True) + + row = 0 + ttk.Label(frame, text="Host:").grid(row=row, column=0, sticky="e", padx=(0, 8), pady=4) + self._host_var = tk.StringVar(value=host) + self._host_entry = ttk.Entry(frame, textvariable=self._host_var, width=28) + self._host_entry.grid(row=row, column=1, columnspan=2, sticky="ew", pady=4) + + row += 1 + ttk.Label(frame, text="Port:").grid(row=row, column=0, sticky="e", padx=(0, 8), pady=4) + self._port_var = tk.StringVar(value=port) + ttk.Entry(frame, textvariable=self._port_var, width=8).grid( + row=row, column=1, sticky="w", pady=4) + + row += 1 + ttk.Label(frame, text="Username:").grid(row=row, column=0, sticky="e", padx=(0, 8), pady=4) + self._user_var = tk.StringVar(value=user) + ttk.Entry(frame, textvariable=self._user_var, width=28).grid( + row=row, column=1, columnspan=2, sticky="ew", pady=4) + + row += 1 + ttk.Label(frame, text="Password:").grid(row=row, column=0, sticky="e", padx=(0, 8), pady=4) + self._pass_var = tk.StringVar() + ttk.Entry(frame, textvariable=self._pass_var, width=28, show="*").grid( + row=row, column=1, columnspan=2, sticky="ew", pady=4) + + row += 1 + self._status_label = ttk.Label(frame, text="", foreground="#666666") + self._status_label.grid(row=row, column=0, columnspan=3, sticky="w", pady=(8, 4)) + + row += 1 + btn_frame = ttk.Frame(frame) + btn_frame.grid(row=row, column=0, columnspan=3, sticky="e", pady=(8, 0)) + + self._btn_connect = ttk.Button(btn_frame, text="Connect", command=self._do_connect) + self._btn_connect.pack(side=tk.RIGHT, padx=(8, 0)) + + self._btn_cancel = ttk.Button(btn_frame, text="Cancel", command=self._do_cancel) + self._btn_cancel.pack(side=tk.RIGHT) + + self._host_entry.focus_set() + self._dlg.bind("", lambda e: self._do_connect()) + self._dlg.bind("", lambda e: self._do_cancel()) + + self._dlg.update_idletasks() + x = parent.winfo_x() + (parent.winfo_width() - self._dlg.winfo_width()) // 2 + y = parent.winfo_y() + (parent.winfo_height() - self._dlg.winfo_height()) // 2 + self._dlg.geometry(f"+{x}+{y}") + + def _do_connect(self): + host = self._host_var.get().strip() + port = self._port_var.get().strip() + user = self._user_var.get().strip() + password = self._pass_var.get() + + if not host: + self._status_label.config(text="Host is required", foreground="#c00000") + return + + try: + port_int = int(port) + except ValueError: + self._status_label.config(text="Invalid port number", foreground="#c00000") + return + + self._btn_connect.config(state="disabled") + self._status_label.config(text="Connecting...", foreground="#666666") + + self._cancelled = False + self._thread = threading.Thread( + target=self._connect_worker, + args=(host, port_int, user, password), + daemon=True + ) + self._thread.start() + self._poll_connect() + + def _connect_worker(self, host, port, user, password): + try: + result = self._callback(host, port, user, password) + self._result = ("ok", result, host, port, user) + except Exception as e: + self._result = ("error", str(e), host, port, user) + + def _poll_connect(self): + if self._cancelled: + return + if self._thread is not None and self._thread.is_alive(): + self._dlg.after(100, self._poll_connect) + return + + status, *rest = self._result + if status == "ok": + _, host, port, user = rest + _save_last_connection(host, str(port), user) + self._dlg.destroy() + else: + msg = rest[0] + self._status_label.config(text=f"Error: {msg}", foreground="#c00000") + self._btn_connect.config(state="normal") + + def _do_cancel(self): + self._cancelled = True + self._dlg.destroy() + + +def _load_last_connection(): + """Reads last connection params from editor.properties.""" + config = configparser.ConfigParser() + host = "192.168.1.100" + port = "1876" + user = "admin" + try: + if app_state.PROPERTIES_FILE: + config.read(app_state.PROPERTIES_FILE) + host = config.get("Device", "host", fallback=host) + port = config.get("Device", "port", fallback=port) + user = config.get("Device", "user", fallback=user) + except Exception: + pass + return host, port, user + + +def _save_last_connection(host, port, user): + """Saves connection params to editor.properties.""" + config = configparser.ConfigParser() + try: + config.read(app_state.PROPERTIES_FILE) + if "Device" not in config: + config["Device"] = {} + config["Device"]["host"] = host + config["Device"]["port"] = port + config["Device"]["user"] = user + with open(app_state.PROPERTIES_FILE, "w") as f: + config.write(f) + except Exception: + pass diff --git a/dasp.py b/dasp.py new file mode 100644 index 0000000..7f8c540 --- /dev/null +++ b/dasp.py @@ -0,0 +1,952 @@ +""" +DASP (Datagram Authenticated Session Protocol) transport layer. + +Ported from sedonaOntrol/src/sedona/dasp/ Java sources. +Provides reliable, authenticated, full-duplex UDP communication +as the transport layer beneath the Sox application protocol. +""" + +import hashlib +import os +import queue +import random +import socket +import struct +import threading +import time + + +# ──────────────────────────────────────────────────────────────── +# Constants (from DaspConst.java) +# ──────────────────────────────────────────────────────────────── + +VERSION_VAL = 0x0100 +IDEAL_MAX_VAL = 512 +ABS_MAX_VAL = 1024 +RECEIVE_MAX_VAL = 31 +RECEIVE_TIMEOUT_VAL = 30_000 # ms +CONNECT_TIMEOUT_VAL = 10_000 # ms +SEND_RETRY_VAL = 1000 # ms +MAX_RETRY_BACKOFF = 8000 # ms (Ontrol) +ACK_MORE_BYTE_LIMIT = 7 # (Ontrol) +DIGEST_ALGORITHM = "SHA-1" # (Ontrol: also "SHA-256") + +IDEAL_MAX_DEF = 512 +ABS_MAX_DEF = 512 +RECEIVE_MAX_DEF = 31 +RECEIVE_TIMEOUT_DEF = 30_000 # ms + +# Message types +DISCOVER = 0 +HELLO = 1 +CHALLENGE = 2 +AUTHENTICATE = 3 +WELCOME = 4 +KEEPALIVE = 5 +DATAGRAM = 6 +CLOSE = 7 + +# Header field value encodings (low 2 bits of field id) +NIL = 0 +U2 = 1 +STR = 2 +BYTES = 3 + +# Header field ids +HDR_VERSION = 0x05 +HDR_REMOTE_ID = 0x09 +HDR_DIGEST_ALGORITHM = 0x0E +HDR_NONCE = 0x13 +HDR_USERNAME = 0x16 +HDR_DIGEST = 0x1B +HDR_IDEAL_MAX = 0x1D +HDR_ABS_MAX = 0x21 +HDR_ACK = 0x25 +HDR_ACK_MORE = 0x2B +HDR_RECEIVE_MAX = 0x2D +HDR_RECEIVE_TIMEOUT = 0x31 +HDR_ERROR_CODE = 0x35 +HDR_PLATFORM_ID = 0x3A +HDR_MESSAGE_LEN = 0x81 # (Ontrol) + +# Error codes +ERR_INCOMPATIBLE_VERSION = 0xE1 +ERR_BUSY = 0xE2 +ERR_DIGEST_NOT_SUPPORTED = 0xE3 +ERR_NOT_AUTHENTICATED = 0xE4 +ERR_TIMEOUT = 0xE5 + +DEFAULT_PORT = 1876 + + +class DaspException(Exception): + def __init__(self, msg, error_code=-1): + super().__init__(msg) + self.error_code = error_code + + +# ──────────────────────────────────────────────────────────────── +# DaspMsg (from DaspMsg.java) +# ──────────────────────────────────────────────────────────────── + +class DaspMsg: + """Binary encoder/decoder for DASP messages.""" + + def __init__(self): + self.session_id = -1 + self.msg_type = -1 + self.seq_num = -1 + # headers + self.version = -1 + self.remote_id = -1 + self.digest_algorithm = None + self.nonce = None + self.username = None + self.digest = None + self.ideal_max = -1 + self.abs_max = -1 + self.ack = -1 + self.ack_more = None + self.receive_max = -1 + self.receive_timeout = -1 # ms internally + self.error_code = -1 + self.platform_id = None + self.include_message_len = True # (Ontrol) + # payload + self.payload = b"" + + # ── Decode ────────────────────────────────────────────────── + + @staticmethod + def decode(buf, length=None): + if length is None: + length = len(buf) + msg = DaspMsg() + msg._decode(buf, length) + return msg + + def _decode(self, buf, length): + self.session_id = (buf[0] << 8) | buf[1] + self.seq_num = (buf[2] << 8) | buf[3] + self.msg_type = (buf[4] & 0xFF) >> 4 + num_fields = buf[4] & 0x0F + + pos = 5 + for _ in range(num_fields): + fid = buf[pos] & 0xFF + pos += 1 + ftype = fid & 0x03 + + u2_val = 0 + str_val = None + bytes_val = None + + if ftype == NIL: + pass + elif ftype == U2: + u2_val = (buf[pos] << 8) | buf[pos + 1] + pos += 2 + elif ftype == STR: + start = pos + while buf[pos] != 0: + pos += 1 + str_val = buf[start:pos].decode("utf-8") + pos += 1 # skip null terminator + elif ftype == BYTES: + blen = buf[pos] & 0xFF + pos += 1 + bytes_val = bytes(buf[pos:pos + blen]) + pos += blen + + if fid == HDR_VERSION: self.version = u2_val + elif fid == HDR_REMOTE_ID: self.remote_id = u2_val + elif fid == HDR_DIGEST_ALGORITHM: self.digest_algorithm = str_val + elif fid == HDR_NONCE: self.nonce = bytes_val + elif fid == HDR_USERNAME: self.username = str_val + elif fid == HDR_DIGEST: self.digest = bytes_val + elif fid == HDR_IDEAL_MAX: self.ideal_max = u2_val + elif fid == HDR_ABS_MAX: self.abs_max = u2_val + elif fid == HDR_ACK: self.ack = u2_val + elif fid == HDR_ACK_MORE: self.ack_more = bytes_val + elif fid == HDR_RECEIVE_MAX: self.receive_max = u2_val + elif fid == HDR_RECEIVE_TIMEOUT: self.receive_timeout = u2_val * 1000 + elif fid == HDR_ERROR_CODE: self.error_code = u2_val + elif fid == HDR_PLATFORM_ID: self.platform_id = str_val + elif fid == HDR_MESSAGE_LEN: pass # not used after decode + + self.payload = bytes(buf[pos:length]) + + # ── Encode ────────────────────────────────────────────────── + + def encode(self): + buf = bytearray(ABS_MAX_VAL) + length = self._encode_into(buf) + return bytes(buf[:length]) + + def _encode_into(self, buf): + struct.pack_into(">HH", buf, 0, self.session_id & 0xFFFF, self.seq_num & 0xFFFF) + pos = 5 # skip byte 4, will backpatch + + has_ideal_max = self.ideal_max > -1 and self.ideal_max != IDEAL_MAX_DEF + has_abs_max = self.abs_max > -1 and self.abs_max != ABS_MAX_DEF + has_receive_max = self.receive_max > -1 and self.receive_max != RECEIVE_MAX_DEF + has_receive_timeout = self.receive_timeout > -1 and self.receive_timeout != RECEIVE_TIMEOUT_DEF + + num = 0 + + # Ontrol: MESSAGE_LEN first (backpatched) + msg_len_pos = -1 + if self.include_message_len: + num += 1 + buf[pos] = HDR_MESSAGE_LEN + pos += 1 + msg_len_pos = pos + pos += 2 # placeholder for length + + if self.version > -1: + num += 1; buf[pos] = HDR_VERSION; pos += 1 + pos = _put_u2(buf, pos, self.version) + if self.remote_id > -1: + num += 1; buf[pos] = HDR_REMOTE_ID; pos += 1 + pos = _put_u2(buf, pos, self.remote_id) + if self.digest_algorithm is not None: + num += 1; buf[pos] = HDR_DIGEST_ALGORITHM; pos += 1 + pos = _put_str(buf, pos, self.digest_algorithm) + if self.nonce is not None: + num += 1; buf[pos] = HDR_NONCE; pos += 1 + pos = _put_bytes(buf, pos, self.nonce) + if self.username is not None: + num += 1; buf[pos] = HDR_USERNAME; pos += 1 + pos = _put_str(buf, pos, self.username) + if self.digest is not None: + num += 1; buf[pos] = HDR_DIGEST; pos += 1 + pos = _put_bytes(buf, pos, self.digest) + if has_ideal_max: + num += 1; buf[pos] = HDR_IDEAL_MAX; pos += 1 + pos = _put_u2(buf, pos, self.ideal_max) + if has_abs_max: + num += 1; buf[pos] = HDR_ABS_MAX; pos += 1 + pos = _put_u2(buf, pos, self.abs_max) + if self.ack > -1: + num += 1; buf[pos] = HDR_ACK; pos += 1 + pos = _put_u2(buf, pos, self.ack) + if self.ack_more is not None: + num += 1; buf[pos] = HDR_ACK_MORE; pos += 1 + pos = _put_bytes(buf, pos, self.ack_more) + if has_receive_max: + num += 1; buf[pos] = HDR_RECEIVE_MAX; pos += 1 + pos = _put_u2(buf, pos, self.receive_max) + if has_receive_timeout: + num += 1; buf[pos] = HDR_RECEIVE_TIMEOUT; pos += 1 + pos = _put_u2(buf, pos, int(self.receive_timeout / 1000)) + if self.error_code > -1: + num += 1; buf[pos] = HDR_ERROR_CODE; pos += 1 + pos = _put_u2(buf, pos, self.error_code) + if self.platform_id is not None: + num += 1; buf[pos] = HDR_PLATFORM_ID; pos += 1 + pos = _put_str(buf, pos, self.platform_id) + + # backpatch msg_type | num_fields + buf[4] = ((self.msg_type & 0x0F) << 4) | (num & 0x0F) + + # payload + if self.payload: + buf[pos:pos + len(self.payload)] = self.payload + pos += len(self.payload) + + # backpatch MESSAGE_LEN + if msg_len_pos >= 0: + _put_u2(buf, msg_len_pos, pos) + + return pos + + +def _put_u2(buf, pos, val): + buf[pos] = (val >> 8) & 0xFF + buf[pos + 1] = val & 0xFF + return pos + 2 + +def _put_str(buf, pos, val): + encoded = val.encode("ascii") + buf[pos:pos + len(encoded)] = encoded + pos += len(encoded) + buf[pos] = 0 + return pos + 1 + +def _put_bytes(buf, pos, val): + buf[pos] = len(val) & 0xFF + pos += 1 + buf[pos:pos + len(val)] = val + return pos + len(val) + + +# ──────────────────────────────────────────────────────────────── +# ReceiveWindow (from ReceiveWindow.java) +# ──────────────────────────────────────────────────────────────── + +class ReceiveWindow: + """Manages incoming message sequence numbers and acknowledgements.""" + + MAX = RECEIVE_MAX_VAL # 31 + + def __init__(self, session): + self._session = session + self._lock = threading.Lock() + self._seq_num = -1 + self._seq_more = -1 + self._acked = False + + def unacked(self): + with self._lock: + return not self._acked + + def init(self, num): + with self._lock: + if self._seq_num >= 0: + raise RuntimeError("ReceiveWindow already initialized") + self._seq_num = (num - 1) & 0xFFFF + self._seq_more = 1 + self._acked = False + + def set_ack_headers(self, msg): + with self._lock: + payload_len = len(msg.payload) + # room for 5-byte fixed header + 3-byte ack header = 8 + if payload_len + 8 <= self._session.ideal_max: + msg.ack = self._seq_num + self._acked = True + # room for ack_more: +3 bytes = 11 total overhead + if self._seq_more > 0x01 and payload_len + 11 <= self._session.ideal_max: + max_bytes = self._session.ideal_max - payload_len - 10 + msg.ack_more = _to_ack_more(self._seq_more, max_bytes) + + def receive(self, num): + with self._lock: + if num < 0 or num > 0xFFFF: + raise RuntimeError(f"Invalid seq num: {num}") + if self._seq_num < 0: + raise RuntimeError("ReceiveWindow not initialized") + + start = self._seq_num + end = start + self.MAX + + if end <= 0xFFFF: + if num <= start or num > end: + self._acked = False + return False + else: + end -= 0xFFFF + if num <= start and num > end: + self._acked = False + return False + + # next in sequence — slide window + if num == (start + 1) & 0xFFFF: + while True: + self._seq_num = (self._seq_num + 1) & 0xFFFF + self._seq_more = (self._seq_more >> 1) & 0x7FFFFFFF + if (self._seq_more & 0x02) == 0: + break + self._seq_more |= 0x01 + else: + # out of order + bit = 1 << ((num - start) & 0xFFFF) + if self._seq_more & bit: + return False + self._seq_more |= bit + + self._acked = False + return True + + +def _to_ack_more(seq_more, max_bytes): + if seq_more <= 0xFF or max_bytes <= 1: + return bytes([seq_more & 0xFF]) + if seq_more <= 0xFFFF or max_bytes <= 2: + return bytes([(seq_more >> 8) & 0xFF, seq_more & 0xFF]) + if seq_more <= 0xFFFFFF or max_bytes <= 3: + return bytes([(seq_more >> 16) & 0xFF, (seq_more >> 8) & 0xFF, seq_more & 0xFF]) + return bytes([(seq_more >> 24) & 0xFF, (seq_more >> 16) & 0xFF, + (seq_more >> 8) & 0xFF, seq_more & 0xFF]) + + +# ──────────────────────────────────────────────────────────────── +# SendWindow (from SendWindow.java) +# ──────────────────────────────────────────────────────────────── + +class _Packet: + __slots__ = ("seq_num", "payload", "acked", "enqueued_time", + "sent_time", "send_attempts", "next") + + def __init__(self): + self.seq_num = 0 + self.payload = b"" + self.acked = False + self.enqueued_time = 0.0 + self.sent_time = 0.0 + self.send_attempts = 1 + self.next = None + + +class SendWindow: + """Manages the outgoing datagram queue with reliable delivery.""" + + FAST_RETRANSMIT_DUP_ACKS = 2 + + def __init__(self, session): + self._session = session + self._cond = threading.Condition() + self._head = None + self._tail = None + self._is_alive = True + self._size = 0 + self._seq_num = random.randint(0, 0xFFFF) + self.send_size = 8 + self.send_retry = SEND_RETRY_VAL / 1000.0 # seconds + self.max_retry_backoff = MAX_RETRY_BACKOFF / 1000.0 # seconds + self._last_checked_ack = -1 + self._dup_ack_count = 0 + + def size(self): + with self._cond: + return self._size + + def cur_seq_num(self): + return self._seq_num & 0xFFFF + + def kill(self): + with self._cond: + self._is_alive = False + self._cond.notify_all() + + def clear(self): + with self._cond: + self._size = 0 + self._head = None + self._tail = None + self._cond.notify_all() + + def send(self, payload): + p = _Packet() + p.enqueued_time = _ticks() + with self._cond: + p.seq_num = self._seq_num & 0xFFFF + self._seq_num = (self._seq_num + 1) & 0xFFFF + + self._enqueue(p) + p.payload = payload + p.sent_time = _ticks() + self._session._send_raw(self._to_msg(p)) + + def _enqueue(self, p): + with self._cond: + while self._is_alive and (self._full() or self._exceeds_ack_more_limit(p)): + self._cond.wait() + if not self._is_alive: + return + if self._tail is None: + self._head = self._tail = p + else: + self._tail.next = p + self._tail = p + self._size += 1 + self._cond.notify_all() + + def _full(self): + return self._size >= self.send_size + + def _exceeds_ack_more_limit(self, p): + last_ack = self._session.last_ack + if last_ack < 0: + return False + span = (p.seq_num - last_ack) & 0xFFFF + return span > self._session.ack_more_byte_limit + + def _to_msg(self, packet): + msg = DaspMsg() + msg.msg_type = DATAGRAM + msg.session_id = self._session.remote_id + msg.seq_num = packet.seq_num + msg.payload = packet.payload + self._session.receive_window.set_ack_headers(msg) + return msg + + def check_ack_headers(self, msg): + with self._cond: + ack_num = msg.ack + if ack_num < 0: + return + + if self._head is None: + self._track_dup_acks(ack_num) + return + + unack_start = (ack_num + 1) & 0xFFFF + unack_end = (unack_start + 10000) & 0xFFFF + + p = self._head + while p is not None: + if not p.acked and self._is_acked(unack_start, unack_end, msg, p.seq_num): + p.acked = True + if p is self._head and p.acked: + self._head = p.next + self._size -= 1 + p = p.next + + if self._head is None: + self._tail = None + + self._track_dup_acks(ack_num) + self._cond.notify_all() + + def _track_dup_acks(self, ack_num): + if ack_num != self._last_checked_ack: + self._last_checked_ack = ack_num + self._dup_ack_count = 0 + return + self._dup_ack_count += 1 + if self._dup_ack_count < self.FAST_RETRANSMIT_DUP_ACKS: + return + self._dup_ack_count = 0 + if self._head is not None and not self._head.acked: + self._resend(self._head) + + @staticmethod + def _is_acked(unack_start, unack_end, msg, seq_num): + if unack_start <= unack_end: + if seq_num < unack_start or seq_num > unack_end: + return True + else: + if seq_num < unack_start and seq_num > unack_end: + return True + + ack_more = msg.ack_more + if ack_more is not None: + diff = (seq_num - msg.ack) & 0xFFFF + index = len(ack_more) - (diff >> 3) - 1 + if 0 <= index < len(ack_more): + ack_more_byte = ack_more[index] & 0xFF + mask = 1 << (diff & 0x07) + if ack_more_byte & mask: + return True + return False + + def send_retries(self): + with self._cond: + now = _ticks() + oldest = 0.0 + p = self._head + while p is not None: + if not p.acked: + backoff_shift = min(p.send_attempts - 1, 32) + interval = min(self.send_retry * (1 << backoff_shift), + self.max_retry_backoff) + if now - p.sent_time >= interval: + self._resend(p) + oldest = max(oldest, now - p.enqueued_time) + p = p.next + return oldest + + def _resend(self, p): + p.sent_time = _ticks() + p.send_attempts += 1 + self._session.num_retries += 1 + self._session._send_raw(self._to_msg(p)) + + +# ──────────────────────────────────────────────────────────────── +# DaspSession (from DaspSession.java) +# ──────────────────────────────────────────────────────────────── + +class DaspSession: + """ + Client endpoint of a DASP session. Handles the handshake, + keepalive, reliable send/receive, and session close. + """ + + def __init__(self, dasp_socket, session_id, host, port, **options): + self.socket = dasp_socket + self.id = session_id + self.host = host + self.port = port + self.remote_id = -1 + self.is_closed = False + self.close_cause = "???" + self.user = "" + self.password = "" + + self.ideal_max = options.get("ideal_max", IDEAL_MAX_DEF) + self.abs_max = options.get("abs_max", ABS_MAX_DEF) + self.receive_timeout = options.get("receive_timeout", RECEIVE_TIMEOUT_VAL) / 1000.0 # seconds + self.connect_timeout = options.get("connect_timeout", CONNECT_TIMEOUT_VAL) / 1000.0 # seconds + self.ack_more_byte_limit = options.get("ack_more_byte_limit", ACK_MORE_BYTE_LIMIT) + self.digest_algorithm = options.get("digest_algorithm", DIGEST_ALGORITHM) + self.include_message_len = options.get("include_message_len", True) + + self.receive_window = ReceiveWindow(self) + self.send_window = SendWindow(self) + self.send_window.send_retry = options.get("send_retry", SEND_RETRY_VAL) / 1000.0 + self.send_window.max_retry_backoff = options.get("max_retry_backoff", MAX_RETRY_BACKOFF) / 1000.0 + + self._receive_queue = queue.Queue(maxsize=2000) + self.remote_receive_max = RECEIVE_MAX_DEF + self.last_ack = -1 + self.last_send = _ticks() + self.last_receive = _ticks() + self.num_sent = 0 + self.num_received = 0 + self.num_retries = 0 + self._nonce = None + self._close_lock = threading.Lock() + self.on_close = None # callback + + # ── Client handshake ──────────────────────────────────────── + + def connect(self): + res = self._hello() + + if res.msg_type == CLOSE: + raise DaspException("Connection denied", res.error_code) + elif res.msg_type == CHALLENGE: + self.remote_id = res.remote_id + welcome = self._authenticate(res) + elif res.msg_type == WELCOME: + self.remote_id = res.remote_id + welcome = res + else: + raise DaspException(f"Unexpected response to hello: {res.msg_type}") + + if welcome.msg_type == CLOSE: + raise DaspException("Connection denied", welcome.error_code) + elif welcome.msg_type == WELCOME: + self._tune(welcome) + else: + raise DaspException(f"Expected welcome, not {welcome.msg_type}") + + def _hello(self): + msg = DaspMsg() + msg.msg_type = HELLO + msg.session_id = 0xFFFF + msg.seq_num = self.send_window.cur_seq_num() + msg.remote_id = self.id + msg.version = VERSION_VAL + msg.ideal_max = self.ideal_max + msg.abs_max = self.abs_max + msg.receive_max = RECEIVE_MAX_VAL + msg.receive_timeout = int(self.receive_timeout * 1000) + + timeout = max(self.connect_timeout / 3, 1.0) + for _ in range(3): + self._send_raw(msg) + try: + res = self._receive_queue.get(timeout=timeout) + return res + except queue.Empty: + continue + raise DaspException("No response from hello") + + def _authenticate(self, challenge): + algorithm = challenge.digest_algorithm or "SHA-1" + if algorithm not in ("SHA-1", "SHA-256"): + self.close(ERR_DIGEST_NOT_SUPPORTED, "digest not supported") + raise DaspException("Unsupported algorithm", ERR_DIGEST_NOT_SUPPORTED) + + # credentials: always SHA-1 of "user:pass" + cred = hashlib.sha1(f"{self.user}:{self.password}".encode("utf-8")).digest() + + # outer digest uses negotiated algorithm + algo_name = "sha1" if algorithm == "SHA-1" else "sha256" + md = hashlib.new(algo_name) + md.update(cred) + md.update(challenge.nonce) + digest_bytes = md.digest() + + msg = DaspMsg() + msg.msg_type = AUTHENTICATE + msg.session_id = self.remote_id + msg.seq_num = self.send_window.cur_seq_num() + msg.username = self.user + msg.digest = digest_bytes + + timeout = max(self.connect_timeout / 3, 1.0) + for _ in range(3): + self._send_raw(msg) + try: + res = self._receive_queue.get(timeout=timeout) + if res.msg_type != CHALLENGE: + return res + # drop duplicate challenge from multiple hellos + except queue.Empty: + continue + raise DaspException("No response from authenticate") + + def _tune(self, msg): + remote_ideal = msg.ideal_max if msg.ideal_max > -1 else IDEAL_MAX_DEF + remote_abs = msg.abs_max if msg.abs_max > -1 else ABS_MAX_DEF + remote_recv_max = msg.receive_max if msg.receive_max > -1 else RECEIVE_MAX_DEF + remote_timeout = (msg.receive_timeout if msg.receive_timeout > -1 + else RECEIVE_TIMEOUT_DEF) / 1000.0 + + self.ideal_max = min(remote_ideal, self.ideal_max) + self.abs_max = min(remote_abs, self.abs_max) + self.remote_receive_max = remote_recv_max + self.receive_timeout = max(remote_timeout, self.receive_timeout) + self.send_window.send_size = min(self.remote_receive_max, self.ack_more_byte_limit) + self.receive_window.init(msg.seq_num) + + # ── Send / Receive ────────────────────────────────────────── + + def send(self, payload): + if self.is_closed: + raise DaspException(f"Session closed: {self.close_cause}") + self.send_window.send(payload) + + def receive(self, timeout=1.0): + if self.is_closed: + raise DaspException(f"Session closed: {self.close_cause}") + try: + msg = self._receive_queue.get(timeout=timeout) + except queue.Empty: + return None + if self.is_closed: + raise DaspException(f"Session closed: {self.close_cause}") + if not isinstance(msg, DaspMsg): + return msg + if msg.msg_type != DATAGRAM: + raise DaspException(f"Invalid message type received: {msg.msg_type}") + return msg + + def _send_raw(self, msg): + msg.include_message_len = self.include_message_len + self.last_send = _ticks() + self.num_sent += 1 + self.socket._send_msg(self, msg) + + # ── Dispatch (called on receiver thread) ──────────────────── + + def dispatch(self, msg): + self.num_received += 1 + self.last_receive = _ticks() + + if msg.ack >= 0: + self.last_ack = msg.ack + + self.send_window.check_ack_headers(msg) + + if msg.msg_type == DATAGRAM: + if not self.receive_window.receive(msg.seq_num): + return + self._enqueue(msg) + elif msg.msg_type in (CHALLENGE, WELCOME): + self._enqueue(msg) + elif msg.msg_type == CLOSE: + self._enqueue(msg) + self._shutdown(f"remote close (error={msg.error_code})") + elif msg.msg_type == KEEPALIVE: + pass # lastReceive already updated + + def _enqueue(self, msg): + try: + self._receive_queue.put_nowait(msg) + except queue.Full: + self.close(ERR_TIMEOUT, "receive queue full") + + # ── Housekeeping (called every 100ms) ─────────────────────── + + def housekeeping(self): + if self.is_closed: + return + + if _ticks() - self.last_receive > self.receive_timeout: + self.close(ERR_TIMEOUT, "receive timeout") + return + + oldest = self.send_window.send_retries() + if oldest > self.receive_timeout: + self.close(ERR_TIMEOUT, "unacked send timeout") + return + + if (self.receive_window.unacked() or + _ticks() - self.last_send > self.receive_timeout / 3): + self._keepalive() + + def _keepalive(self): + msg = DaspMsg() + msg.msg_type = KEEPALIVE + msg.session_id = self.remote_id + msg.seq_num = 0xFFFF + self.receive_window.set_ack_headers(msg) + self._send_raw(msg) + + # ── Close ─────────────────────────────────────────────────── + + def close(self, error_code=-1, cause="close()"): + with self._close_lock: + if self.is_closed: + return + close_msg = DaspMsg() + close_msg.msg_type = CLOSE + close_msg.session_id = self.remote_id if self.remote_id >= 0 else 0xFFFF + close_msg.seq_num = 0xFFFF + if error_code >= 0: + close_msg.error_code = error_code + self._send_raw(close_msg) + if error_code < 0: + time.sleep(0.05) + self._send_raw(close_msg) + self._shutdown(cause) + + def _shutdown(self, cause): + with self._close_lock: + if self.is_closed: + return + self.is_closed = True + self.close_cause = cause + self.send_window.kill() + # wake anyone blocked on receive + try: + self._receive_queue.put_nowait(None) + except queue.Full: + pass + if self.on_close: + try: + self.on_close(self) + except Exception: + pass + self.socket._free_session(self) + + +# ──────────────────────────────────────────────────────────────── +# DaspSocket (from DaspSocket.java + DaspSocketInterface.java) +# ──────────────────────────────────────────────────────────────── + +class DaspSocket: + """ + Manages a UDP socket shared by DASP sessions. + Provides receiver and housekeeping background threads. + """ + + def __init__(self, port=-1): + self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self._sock.settimeout(1.0) + if port > 0: + self._sock.bind(("", port)) + else: + self._sock.bind(("", 0)) + + self._sessions = {} # id -> DaspSession + self._sessions_lock = threading.Lock() + self._is_alive = True + self._rand = random.Random() + self._send_lock = threading.Lock() + + self._receiver_thread = threading.Thread( + target=self._receiver_loop, name="DASP-Receiver", daemon=True) + self._housekeeping_thread = threading.Thread( + target=self._housekeeping_loop, name="DASP-Housekeeping", daemon=True) + self._receiver_thread.start() + self._housekeeping_thread.start() + + @property + def port(self): + return self._sock.getsockname()[1] + + @property + def is_closed(self): + return not self._is_alive + + def connect(self, host, port, user, password, **options): + if not self._is_alive: + raise DaspException("Socket is closed") + session = self._alloc_session(host, port, **options) + session.user = user + session.password = password + session.connect() + return session + + def close(self): + if not self._is_alive: + return + self._is_alive = False + with self._sessions_lock: + for s in list(self._sessions.values()): + try: + s.close(-1, "socket closed") + except Exception: + pass + try: + self._sock.close() + except Exception: + pass + + # ── Session management ────────────────────────────────────── + + def _alloc_session(self, host, port, **options): + with self._sessions_lock: + while True: + sid = self._rand.randint(0, 0xFFFE) + if sid not in self._sessions: + break + session = DaspSession(self, sid, host, port, **options) + self._sessions[sid] = session + return session + + def _free_session(self, session): + with self._sessions_lock: + self._sessions.pop(session.id, None) + + # ── Network I/O ───────────────────────────────────────────── + + def _send_msg(self, session, msg): + data = msg.encode() + with self._send_lock: + try: + self._sock.sendto(data, (session.host, session.port)) + except OSError: + pass + + def _receiver_loop(self): + buf = bytearray(ABS_MAX_VAL + 64) + while self._is_alive: + try: + nbytes, addr = self._sock.recvfrom_into(buf) + if nbytes < 5: + continue + msg = DaspMsg.decode(buf, nbytes) + self._dispatch(msg, addr) + except socket.timeout: + continue + except OSError: + if self._is_alive: + continue + break + + def _dispatch(self, msg, addr): + session_id = msg.session_id + with self._sessions_lock: + session = self._sessions.get(session_id) + if session is None: + return + if session.host != addr[0] or session.port != addr[1]: + return + session.dispatch(msg) + + def _housekeeping_loop(self): + while self._is_alive: + time.sleep(0.1) + with self._sessions_lock: + sessions = list(self._sessions.values()) + for s in sessions: + try: + s.housekeeping() + except Exception: + pass + + +# ──────────────────────────────────────────────────────────────── +# Utility +# ──────────────────────────────────────────────────────────────── + +def _ticks(): + """Monotonic time in seconds (float).""" + return time.monotonic() diff --git a/device_session.py b/device_session.py new file mode 100644 index 0000000..73941e8 --- /dev/null +++ b/device_session.py @@ -0,0 +1,334 @@ +""" +DeviceSession — bridges SoxClient to the editor's ET.Element-based UI. + +Handles lazy loading, DOM synthesis from cached SoxComponents, +COV event routing, and subscription lifecycle. +""" + +import queue +import xml.etree.ElementTree as ET + +import app_state +import schema +from sox import SoxClient, SoxComponent, TREE, CONFIG, RUNTIME, LINKS, ALL + + +class DeviceSession: + """High-level session manager for a live Sedona device connection.""" + + def __init__(self): + self.sox_client = None + self._event_queue = queue.Queue() + self._current_subs = set() # comp_ids we subscribed for the open wiresheet + + # ── Connect / Disconnect ──────────────────────────────────── + + def connect(self, host, port, user, password): + """DASP connect → read schema → load root. Called from background thread.""" + self.sox_client = SoxClient(host, int(port), user, password) + self.sox_client.on_event = self._on_sox_event + self.sox_client.connect() + + parts = self.sox_client.read_schema() + vi = self.sox_client.read_version() + + self._load_slot_registry(parts) + + root_comp = self.sox_client.load(0) + return root_comp + + def disconnect(self): + if self.sox_client is None: + return + try: + if self._current_subs: + comps = [self.sox_client.cache_get(cid) + for cid in self._current_subs] + comps = [c for c in comps if c is not None] + if comps: + self.sox_client.unsubscribe(comps, ALL) + except Exception: + pass + self._current_subs.clear() + try: + self.sox_client.close() + except Exception: + pass + self.sox_client = None + + @property + def is_connected(self): + return self.sox_client is not None and not self.sox_client.is_closed + + # ── Schema → Slot Registry ────────────────────────────────── + + def _load_slot_registry(self, parts): + """Builds slot_registry from kit manifests for the device's schema.""" + registry = {} + for kit_id, part in enumerate(parts): + kit_name = part.name + checksum = part.checksum + manifest = schema.find_manifest(kit_name, checksum) + if manifest is None: + continue + self._register_kit_types(registry, kit_id, kit_name, manifest) + self.sox_client.slot_registry = registry + + def _register_kit_types(self, registry, kit_id, kit_name, manifest): + root = manifest + for type_elem in root.iter("type"): + type_name = type_elem.get("name", "") + type_id_val = int(type_elem.get("id", "-1")) + if type_id_val < 0: + continue + slots = [] + for slot_elem in type_elem.iter("slot"): + slot_name = slot_elem.get("name", "") + slot_id = int(slot_elem.get("id", "-1")) + slot_type = slot_elem.get("type", "int") + flags_str = slot_elem.get("flags", "") + is_config = "c" in flags_str or "p" in flags_str + is_runtime = "r" in flags_str + is_action = "a" in flags_str + is_operator = "o" in flags_str + is_as_str = slot_type == "Str" or slot_elem.get("asStr") == "true" + slots.append({ + "name": slot_name, + "id": slot_id, + "type_name": slot_type, + "is_config": is_config, + "is_runtime": is_runtime, + "is_action": is_action, + "is_operator": is_operator, + "is_as_str": is_as_str, + "is_prop": not is_action, + }) + registry[(kit_id, type_id_val)] = slots + full_name = f"{kit_name}::{type_name}" + # update type_name on any cached component of this type + for c in self.sox_client.cache: + if c is not None and c.kit_id == kit_id and c.type_id == type_id_val: + c.type_name = full_name + + # ── Lazy Loading ──────────────────────────────────────────── + + def load_children(self, comp_id): + """Loads direct children of a component (cache-aware). Returns list of SoxComponent.""" + parent = self.sox_client.cache_get(comp_id) + if parent is None: + parent = self.sox_client.load(comp_id) + if parent is None: + return [] + uncached = [cid for cid in parent.children + if self.sox_client.cache_get(cid) is None] + if uncached: + self.sox_client.load_ids(uncached, checked=False) + return [self.sox_client.cache_get(cid) + for cid in parent.children + if self.sox_client.cache_get(cid) is not None] + + # ── Wiresheet Open / Close ────────────────────────────────── + + def open_wiresheet(self, folder_comp_id): + """Full wiresheet load: children, links, remote comps, config+runtime, subscribe. + + Returns list of SoxComponent for the folder's children. + """ + children = self.load_children(folder_comp_id) + if not children: + return [] + + all_link_remote_ids = set() + for child in children: + links = self.sox_client.read_links(child.id) + child.links = links + for lnk in links: + all_link_remote_ids.add(lnk.from_comp_id) + all_link_remote_ids.add(lnk.to_comp_id) + + child_ids = {c.id for c in children} + child_ids.add(folder_comp_id) + remote_ids = [rid for rid in all_link_remote_ids + if rid not in child_ids + and self.sox_client.cache_get(rid) is None] + if remote_ids: + self.sox_client.load_ids(remote_ids, checked=False) + + self.sox_client.update(children, CONFIG | RUNTIME) + + folder_comp = self.sox_client.cache_get(folder_comp_id) + to_subscribe = [folder_comp] + children if folder_comp else children + to_subscribe = [c for c in to_subscribe if c is not None] + try: + self.sox_client.subscribe(to_subscribe, ALL) + except Exception: + pass + + self._current_subs = {c.id for c in to_subscribe} + return children + + def close_wiresheet(self): + """Unsubscribe from previously subscribed components.""" + if not self._current_subs or not self.is_connected: + self._current_subs.clear() + return + comps = [self.sox_client.cache_get(cid) + for cid in self._current_subs] + comps = [c for c in comps if c is not None] + if comps: + try: + self.sox_client.unsubscribe(comps, ALL) + except Exception: + pass + self._current_subs.clear() + + # ── DOM Synthesis ─────────────────────────────────────────── + + def build_dom_tree(self, root_comp): + """Builds an ET.Element tree from cached SoxComponents, for the editor UI. + + Only builds from what is currently in the cache (lazy-loaded components). + Returns a root element. + """ + sax_root = ET.Element("sedonaApp") + schema_elem = ET.SubElement(sax_root, "schema") + if self.sox_client and self.sox_client.schema_parts: + for part in self.sox_client.schema_parts: + ET.SubElement(schema_elem, "kit", + name=part.name, checksum=str(part.checksum)) + + app_elem = ET.SubElement(sax_root, "app") + if root_comp is not None: + self._build_comp_element(app_elem, root_comp) + + return sax_root + + def _build_comp_element(self, parent_elem, sox_comp): + type_qname = sox_comp.type_name + attribs = { + "name": sox_comp.name, + "id": str(sox_comp.id), + "type": type_qname, + } + comp_elem = ET.SubElement(parent_elem, "comp", **attribs) + + slots = self.sox_client.slot_registry.get( + (sox_comp.kit_id, sox_comp.type_id)) + if slots: + for slot in slots: + if slot.get("is_action"): + continue + name = slot["name"] + if name in sox_comp.values: + val = sox_comp.values[name] + ET.SubElement(comp_elem, "prop", + name=name, val=str(val)) + + for child_id in sox_comp.children: + child_comp = self.sox_client.cache_get(child_id) + if child_comp is not None: + self._build_comp_element(comp_elem, child_comp) + + return comp_elem + + def build_links_elements(self, sax_root): + """Adds elements under the app element from cached link data.""" + app_elem = sax_root.find("app") + if app_elem is None: + return + for c in self.sox_client.cache: + if c is None: + continue + for lnk in c.links: + from_comp = self.sox_client.cache_get(lnk.from_comp_id) + to_comp = self.sox_client.cache_get(lnk.to_comp_id) + if from_comp is None or to_comp is None: + continue + + from_slot_name = self._slot_name_by_id( + from_comp.kit_id, from_comp.type_id, lnk.from_slot_id) + to_slot_name = self._slot_name_by_id( + to_comp.kit_id, to_comp.type_id, lnk.to_slot_id) + + from_path = self._comp_path(lnk.from_comp_id) + to_path = self._comp_path(lnk.to_comp_id) + + if from_path and to_path and from_slot_name and to_slot_name: + ET.SubElement(app_elem, "link", + **{"from": f"{from_path}.{from_slot_name}", + "to": f"{to_path}.{to_slot_name}"}) + + def _slot_name_by_id(self, kit_id, type_id_val, slot_id): + slots = self.sox_client.slot_registry.get((kit_id, type_id_val)) + if slots: + for s in slots: + if s["id"] == slot_id: + return s["name"] + return None + + def _comp_path(self, comp_id): + parts = [] + cid = comp_id + while cid >= 0: + c = self.sox_client.cache_get(cid) + if c is None: + break + if c.id == 0: + break + parts.append(c.name) + cid = c.parent + if not parts: + return None + parts.reverse() + return "/".join(parts) + + # ── Update DOM from COV ───────────────────────────────────── + + def update_element_from_cache(self, comp_id, sax_root): + """Updates a single comp element in the DOM from cached SoxComponent values.""" + sox_comp = self.sox_client.cache_get(comp_id) + if sox_comp is None: + return None + + comp_elem = self._find_element_by_id(sax_root, comp_id) + if comp_elem is None: + return None + + slots = self.sox_client.slot_registry.get( + (sox_comp.kit_id, sox_comp.type_id)) + if slots: + existing_props = {p.get("name"): p for p in comp_elem.findall("prop")} + for slot in slots: + if slot.get("is_action"): + continue + name = slot["name"] + if name in sox_comp.values: + val = str(sox_comp.values[name]) + if name in existing_props: + existing_props[name].set("val", val) + else: + ET.SubElement(comp_elem, "prop", name=name, val=val) + return comp_elem + + @staticmethod + def _find_element_by_id(root, comp_id): + target = str(comp_id) + for elem in root.iter("comp"): + if elem.get("id") == target: + return elem + return None + + # ── Event queue (thread-safe) ─────────────────────────────── + + def _on_sox_event(self, comp_id, what_char): + self._event_queue.put((comp_id, what_char)) + + def poll_events(self): + """Drains the event queue. Returns list of (comp_id, what_char) tuples.""" + events = [] + try: + while True: + events.append(self._event_queue.get_nowait()) + except queue.Empty: + pass + return events diff --git a/live_hooks.py b/live_hooks.py new file mode 100644 index 0000000..5efbcd6 --- /dev/null +++ b/live_hooks.py @@ -0,0 +1,251 @@ +""" +Write-through interception layer for live device mode. + +Checks app_state.live_mode and routes mutations through DeviceSession/SoxClient. +In offline mode, falls through to the original DOM-only code paths. +""" + +import app_state +from sox import type_id_for_slot + + +def is_live(): + return (app_state.live_mode + and app_state.device_session is not None + and app_state.device_session.is_connected) + + +# ── Property Write ────────────────────────────────────────────── + +def live_set_prop_value(element, slot_name, value_str): + """In live mode, sends a Sox write to the device. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + comp_id = _comp_id(element) + if comp_id is None: + return False + + slot_info = _find_slot(ds, comp_id, slot_name) + if slot_info is None: + return False + + type_id = type_id_for_slot(slot_info["type_name"]) + coerced = _coerce_value(type_id, value_str) + + try: + ds.sox_client.write_prop(comp_id, slot_info["id"], type_id, coerced) + return True + except Exception as e: + app_state.set_status(f"Write failed: {e}", warn=True) + return False + + +# ── Add Component ─────────────────────────────────────────────── + +def live_add_component(parent_element, kit_name, type_name, comp_name, config_values=None): + """In live mode, sends a Sox add to the device. Returns new comp_id or None.""" + if not is_live(): + return None + + ds = app_state.device_session + parent_id = _comp_id(parent_element) + if parent_id is None: + return None + + kit_id = _kit_id(ds, kit_name) + type_id_val = _type_id(ds, kit_id, type_name) + if kit_id is None or type_id_val is None: + return None + + sox_config = [] + if config_values: + for tid, val in config_values: + sox_config.append((tid, val)) + + try: + new_comp = ds.sox_client.add(parent_id, kit_id, type_id_val, comp_name, sox_config) + return new_comp.id + except Exception as e: + app_state.set_status(f"Add failed: {e}", warn=True) + return None + + +# ── Delete Component ──────────────────────────────────────────── + +def live_delete_component(element): + """In live mode, sends a Sox delete to the device. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + comp_id = _comp_id(element) + if comp_id is None: + return False + + try: + ds.sox_client.delete(comp_id) + return True + except Exception as e: + app_state.set_status(f"Delete failed: {e}", warn=True) + return False + + +# ── Rename Component ──────────────────────────────────────────── + +def live_rename_component(element, new_name): + """In live mode, sends a Sox rename to the device. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + comp_id = _comp_id(element) + if comp_id is None: + return False + + try: + ds.sox_client.rename(comp_id, new_name) + return True + except Exception as e: + app_state.set_status(f"Rename failed: {e}", warn=True) + return False + + +# ── Link / Unlink ─────────────────────────────────────────────── + +def live_add_link(from_element, from_slot_name, to_element, to_slot_name): + """In live mode, sends a Sox link command. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + from_id = _comp_id(from_element) + to_id = _comp_id(to_element) + if from_id is None or to_id is None: + return False + + from_slot = _find_slot(ds, from_id, from_slot_name) + to_slot = _find_slot(ds, to_id, to_slot_name) + if from_slot is None or to_slot is None: + return False + + try: + ds.sox_client.link(from_id, from_slot["id"], to_id, to_slot["id"]) + return True + except Exception as e: + app_state.set_status(f"Link failed: {e}", warn=True) + return False + + +def live_delete_link(from_element, from_slot_name, to_element, to_slot_name): + """In live mode, sends a Sox unlink command. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + from_id = _comp_id(from_element) + to_id = _comp_id(to_element) + if from_id is None or to_id is None: + return False + + from_slot = _find_slot(ds, from_id, from_slot_name) + to_slot = _find_slot(ds, to_id, to_slot_name) + if from_slot is None or to_slot is None: + return False + + try: + ds.sox_client.unlink(from_id, from_slot["id"], to_id, to_slot["id"]) + return True + except Exception as e: + app_state.set_status(f"Unlink failed: {e}", warn=True) + return False + + +# ── Invoke Action ─────────────────────────────────────────────── + +def live_invoke(element, slot_name, arg=None): + """In live mode, sends a Sox invoke. Returns True on success.""" + if not is_live(): + return False + + ds = app_state.device_session + comp_id = _comp_id(element) + if comp_id is None: + return False + + slot_info = _find_slot(ds, comp_id, slot_name) + if slot_info is None: + return False + + try: + ds.sox_client.invoke(comp_id, slot_info["id"]) + return True + except Exception as e: + app_state.set_status(f"Invoke failed: {e}", warn=True) + return False + + +# ── Helpers ───────────────────────────────────────────────────── + +def _comp_id(element): + cid = element.get("id") + if cid is None: + return None + try: + return int(cid) + except ValueError: + return None + + +def _find_slot(ds, comp_id, slot_name): + comp = ds.sox_client.cache_get(comp_id) + if comp is None: + return None + slots = ds.sox_client.slot_registry.get((comp.kit_id, comp.type_id)) + if slots is None: + return None + for s in slots: + if s["name"] == slot_name: + return s + return None + + +def _kit_id(ds, kit_name): + if ds.sox_client.schema_parts: + for i, part in enumerate(ds.sox_client.schema_parts): + if part.name == kit_name: + return i + return None + + +def _type_id(ds, kit_id, type_name): + if kit_id is None: + return None + for (kid, tid), slots in ds.sox_client.slot_registry.items(): + if kid == kit_id: + comp = ds.sox_client.cache_get(0) + if comp and comp.type_name.endswith(f"::{type_name}"): + return tid + # fallback: search all cached components for a matching type_name + for c in ds.sox_client.cache: + if c is not None and c.kit_id == kit_id and c.type_name.endswith(f"::{type_name}"): + return c.type_id + return None + + +def _coerce_value(type_id, value_str): + from sox import BOOL_ID, BYTE_ID, SHORT_ID, INT_ID, LONG_ID, FLOAT_ID, DOUBLE_ID, STR_ID + if type_id == BOOL_ID: + return value_str.lower() in ("true", "1", "yes") + elif type_id in (BYTE_ID, SHORT_ID, INT_ID): + return int(value_str) + elif type_id == LONG_ID: + return int(value_str) + elif type_id == FLOAT_ID: + return float(value_str) + elif type_id == DOUBLE_ID: + return float(value_str) + elif type_id == STR_ID: + return str(value_str) + return value_str diff --git a/main.py b/main.py index d4f4a93..af180b9 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,11 @@ import sys import tkinter as tk from tkinter import filedialog, font as tkfont, messagebox, ttk +import threading + import app_state +import connect_dialog +import device_session import link_view import manage_kits import palette @@ -393,6 +397,9 @@ def _clear_restoring_flag(): def do_undo(event=None): + if app_state.live_mode: + app_state.set_status("Undo is not available in live mode", warn=True) + return "break" label = undo.undo() if label is None: return "break" @@ -401,6 +408,9 @@ def do_undo(event=None): def do_redo(event=None): + if app_state.live_mode: + app_state.set_status("Redo is not available in live mode", warn=True) + return "break" label = undo.redo() if label is None: return "break" @@ -409,10 +419,32 @@ def do_redo(event=None): def do_save(event=None): + if app_state.live_mode: + do_download_to_sax() + return "break" sax_file.execute_direct_overwrite_save() return "break" +def do_download_to_sax(): + """In live mode, saves the current device DOM snapshot as a .sax file.""" + if app_state.xml_root_element is None: + return + path = filedialog.asksaveasfilename( + title="Download to SAX", + defaultextension=".sax", + filetypes=[("Sedona SAX", "*.sax"), ("All files", "*.*")], + initialfile=f"{app_state.device_host}.sax" + ) + if not path: + return + try: + sax_file.write_sax_to_path(path) + app_state.set_status(f"Downloaded to {os.path.basename(path)}") + except Exception as e: + messagebox.showerror("Download Failed", str(e)) + + NL = chr(10) # newline, for the multi-line message boxes @@ -578,6 +610,23 @@ def on_tree_select(event): app_state.current_sheet_parent = element record_visit(VIEW_SHEET, element) + if app_state.live_mode and app_state.device_session and element is not None: + comp_id_str = element.get("id") + if comp_id_str is not None: + ds = app_state.device_session + ds.close_wiresheet() + try: + ds.open_wiresheet(int(comp_id_str)) + sax_root = ds.build_dom_tree(ds.sox_client.cache_get(0)) + ds.build_links_elements(sax_root) + app_state.xml_root_element = sax_root + sax_file.refresh_treeview_from_dom(sax_root) + # re-find the element after DOM rebuild + element = app_state.tree_element_map.get(selected_id) + app_state.current_sheet_parent = element + except Exception: + pass + if element is not None: components_to_render = [child for child in element if child.tag == "comp"] else: @@ -1290,7 +1339,18 @@ def attach_tree_scrollbars(container, tree): # Resizing, and opening or closing a branch, both change what the widest row is. tree.bind("", lambda event: fit_tree_column(tree), add="+") - tree.bind("<>", lambda event: tree.after_idle(fit_tree_column, tree), add="+") + def _on_tree_open(event): + tree.after_idle(fit_tree_column, tree) + if app_state.live_mode: + sel = tree.selection() + if sel: + elem = app_state.tree_element_map.get(sel[0]) + if elem is not None: + cid = elem.get("id") + if cid is not None: + _live_tree_expand(int(cid)) + + tree.bind("<>", _on_tree_open, add="+") tree.bind("<>", lambda event: tree.after_idle(fit_tree_column, tree), add="+") @@ -1838,6 +1898,143 @@ def verify_startup_configurations(): wiresheet.render_components(app_state.active_components_list) +# ── Live device connection ────────────────────────────────────── + +_cov_poll_id = None + + +def do_device_connect(): + """Opens the connect dialog and initiates a device connection.""" + connect_dialog.ConnectDialog(app_state.window, _connect_to_device) + + +def _connect_to_device(host, port, user, password): + """Called from the connect dialog's background thread.""" + ds = device_session.DeviceSession() + root_comp = ds.connect(host, port, user, password) + + app_state.device_session = ds + app_state.device_host = host + app_state.device_port = port + app_state.live_mode = True + app_state.connection_status = "connected" + + app_state.window.after(0, lambda: _on_connected(ds, root_comp)) + return root_comp + + +def _on_connected(ds, root_comp): + """Main-thread callback after successful connect.""" + app_state.window.title(app_state.base_window_title_live()) + menu_device.entryconfig("Connect...", state="disabled") + menu_device.entryconfig("Disconnect", state="normal") + menu_device.entryconfig("Download to SAX...", state="normal") + app_state.set_status(f"Connected to {app_state.device_host}:{app_state.device_port}") + + sax_root = ds.build_dom_tree(root_comp) + app_state.xml_root_element = sax_root + app_state.current_file_path = f"device://{app_state.device_host}" + + _load_live_schema(ds) + sax_file.refresh_treeview_from_dom(sax_root) + + undo.reset() + app_state.mark_clean() + _start_cov_polling() + + +def _load_live_schema(ds): + """Loads schema kits from the device's schema parts into the editor registry.""" + app_state.active_schema_kits.clear() + app_state.final_components_registry.clear() + if ds.sox_client and ds.sox_client.schema_parts: + for part in ds.sox_client.schema_parts: + app_state.active_schema_kits[part.name] = f"{part.checksum:08x}" + try: + schema.load_schema_kits_and_manifests() + except Exception: + pass + + +def do_device_disconnect(): + """Disconnects from the live device.""" + global _cov_poll_id + + if _cov_poll_id is not None: + app_state.window.after_cancel(_cov_poll_id) + _cov_poll_id = None + + ds = app_state.device_session + if ds is not None: + ds.disconnect() + + app_state.device_session = None + app_state.live_mode = False + app_state.connection_status = "offline" + app_state.device_host = "" + app_state.device_port = 1876 + + app_state.window.title(app_state.base_window_title()) + menu_device.entryconfig("Connect...", state="normal") + menu_device.entryconfig("Disconnect", state="disabled") + menu_device.entryconfig("Download to SAX...", state="disabled") + app_state.set_status("Disconnected") + + sax_file.reset_to_startup_state() + sax_file.refresh_treeview_from_dom(None) + wiresheet.render_components([]) + + +def _start_cov_polling(): + """Starts the 100ms COV event polling loop on the main thread.""" + global _cov_poll_id + + def poll(): + global _cov_poll_id + ds = app_state.device_session + if ds is None or not ds.is_connected: + _cov_poll_id = None + if app_state.live_mode: + app_state.connection_status = "error" + app_state.set_status("Connection lost", warn=True) + return + + events = ds.poll_events() + if events and app_state.xml_root_element is not None: + needs_repaint = False + for comp_id, what_char in events: + elem = ds.update_element_from_cache(comp_id, app_state.xml_root_element) + if elem is not None: + needs_repaint = True + if needs_repaint: + wiresheet.render_current_sheet() + if properties_panel is not None: + try: + properties_panel.draw() + except Exception: + pass + + _cov_poll_id = app_state.window.after(100, poll) + + _cov_poll_id = app_state.window.after(100, poll) + + +def _live_tree_expand(comp_id): + """Called when user expands a tree node in live mode to lazy-load children.""" + ds = app_state.device_session + if ds is None or not ds.is_connected: + return + + children = ds.load_children(comp_id) + if not children: + return + + sax_root = ds.build_dom_tree(ds.sox_client.cache_get(0)) + ds.build_links_elements(sax_root) + app_state.xml_root_element = sax_root + sax_file.refresh_treeview_from_dom(sax_root) + + # --- UI Setup --- window = tk.Tk() app_state.window = window @@ -1976,6 +2173,13 @@ menu_app_info = tk.Menu(menu_bar, tearoff=0) menu_app_info.add_command(label="App Info", command=show_app_info) menu_bar.add_cascade(label="App", menu=menu_app_info) +menu_device = tk.Menu(menu_bar, tearoff=0) +menu_device.add_command(label="Connect...", command=do_device_connect) +menu_device.add_command(label="Disconnect", command=do_device_disconnect, state="disabled") +menu_device.add_separator() +menu_device.add_command(label="Download to SAX...", command=do_download_to_sax, state="disabled") +menu_bar.add_cascade(label="Device", menu=menu_device) + menu_prefs = tk.Menu(menu_bar, tearoff=0) menu_prefs.add_command(label="Preferences", command=open_preferences_dialog) menu_bar.add_cascade(label="Preferences", menu=menu_prefs) diff --git a/properties.py b/properties.py index f936e06..34bd296 100644 --- a/properties.py +++ b/properties.py @@ -32,6 +32,7 @@ import tkinter as tk from tkinter import font as tkfont, messagebox, ttk import app_state +import live_hooks import schema import undo from sax_dom import (get_component_path, read_meta_property, set_prop_value, @@ -1132,11 +1133,16 @@ class PropertiesPanel: self.status_var.set("No changes to save.") return + live_failures = [] for element, prop_name, _before, after in changes: if prop_name != "meta": set_prop_value(element, prop_name, after) + if live_hooks.is_live(): + if not live_hooks.live_set_prop_value(element, prop_name, after): + live_failures.append(prop_name) - undo.push(undo.SlotValueCommand(list(changes))) + if not app_state.live_mode: + undo.push(undo.SlotValueCommand(list(changes))) app_state.mark_dirty() self.refresh() @@ -1144,10 +1150,19 @@ class PropertiesPanel: self.on_change() count = len(changes) - self.status_var.set( - f"{count} slot{'' if count == 1 else 's'} changed. " - "Ctrl+S writes the file." - ) + if live_failures: + self.status_var.set( + f"Written locally but failed on device: {', '.join(live_failures)}" + ) + elif app_state.live_mode: + self.status_var.set( + f"{count} slot{'' if count == 1 else 's'} written to device." + ) + else: + self.status_var.set( + f"{count} slot{'' if count == 1 else 's'} changed. " + "Ctrl+S writes the file." + ) app_state.set_status( f"{self.comp.get('name', '?')}: {count} slot{'' if count == 1 else 's'} changed." ) diff --git a/sax_file.py b/sax_file.py index 61e0765..cd3cea1 100644 --- a/sax_file.py +++ b/sax_file.py @@ -6,11 +6,13 @@ from tkinter import filedialog, messagebox import xml.etree.ElementTree as ET import app_state +import live_hooks import palette import schema import undo import wiresheet from sax_dom import ( + find_component_by_path, find_parent_element, get_component_path, restore_children, @@ -253,6 +255,18 @@ def execute_direct_overwrite_save(): messagebox.showerror("Save Error", f"Failed to write direct update stream:\n{str(e)}") +def write_sax_to_path(path): + """Writes the current DOM to a .sax file at the given path.""" + if app_state.xml_root_element is None: + raise RuntimeError("No DOM to write") + sync_live_metadata_to_dom() + with open(path, "wb") as f: + f.write(b'\n') + ET.ElementTree(app_state.xml_root_element).write( + f, encoding="utf-8", xml_declaration=False + ) + + def save_file_as(): """Saves the tracking tree document hierarchy targeting an explicit location choice dialog.""" if not app_state.current_file_path or app_state.xml_root_element is None: @@ -754,9 +768,16 @@ def add_component(kit_name, type_name, grid_x, grid_y): reset_to_startup_state() return None - comp_id = allocate_component_id() comp_name = unique_child_name(parent, type_name) + if live_hooks.is_live(): + new_id = live_hooks.live_add_component(parent, kit_name, type_name, comp_name) + if new_id is None: + return None + comp_id = new_id + else: + comp_id = allocate_component_id() + element = ET.Element("comp") element.set("name", comp_name) element.set("id", str(comp_id)) @@ -769,18 +790,19 @@ def add_component(kit_name, type_name, grid_x, grid_y): index = len(list(parent)) parent.append(element) - undo.push(undo.AddComponentCommand( - parent, index, element, - label=f"add {comp_name}" - )) + if not app_state.live_mode: + undo.push(undo.AddComponentCommand( + parent, index, element, + label=f"add {comp_name}" + )) app_state.mark_dirty() refresh_treeview_from_dom(app_state.xml_root_element) wiresheet.render_current_sheet() - # The schema is only reconciled on save, so the placement says so here rather than - # letting the save dialog be the first the user hears of it. - if kit_name in app_state.pending_schema_kits: + if app_state.live_mode: + app_state.set_status(f"{comp_name} ({kit_name}::{type_name}) added to device.") + elif kit_name in app_state.pending_schema_kits: app_state.set_status( f"{comp_name} ({kit_name}::{type_name}) added — '{kit_name}' is not in this " "app's schema and will be declared when you save.", @@ -801,20 +823,31 @@ def remove_link(link_element): if links_elem is None: return + from_ref = link_element.get("from", "") + to_ref = link_element.get("to", "") + + if live_hooks.is_live(): + from_comp_path, from_slot_name = from_ref.rsplit(".", 1) if "." in from_ref else ("", from_ref) + to_comp_path, to_slot_name = to_ref.rsplit(".", 1) if "." in to_ref else ("", to_ref) + from_elem = find_component_by_path(app_state.xml_root_element, from_comp_path) + to_elem = find_component_by_path(app_state.xml_root_element, to_comp_path) + if from_elem is not None and to_elem is not None: + if not live_hooks.live_delete_link(from_elem, from_slot_name, to_elem, to_slot_name): + return + try: index = list(links_elem).index(link_element) links_elem.remove(link_element) except ValueError: return - from_ref = link_element.get("from", "") - to_ref = link_element.get("to", "") from_slot = from_ref.rsplit(".", 1)[-1] if "." in from_ref else from_ref to_slot = to_ref.rsplit(".", 1)[-1] if "." in to_ref else to_ref - undo.push(undo.DeleteLinkCommand( - links_elem, index, link_element, - label=f"delete link {from_slot}→{to_slot}" - )) + if not app_state.live_mode: + undo.push(undo.DeleteLinkCommand( + links_elem, index, link_element, + label=f"delete link {from_slot}→{to_slot}" + )) app_state.selected_link_tag = None app_state.mark_dirty() @@ -830,16 +863,26 @@ def add_link(from_path_slot, to_path_slot): if links_elem is None: links_elem = ET.SubElement(app_state.xml_root_element, "links") + if live_hooks.is_live(): + from_comp_path, from_slot_name = from_path_slot.rsplit(".", 1) if "." in from_path_slot else ("", from_path_slot) + to_comp_path, to_slot_name = to_path_slot.rsplit(".", 1) if "." in to_path_slot else ("", to_path_slot) + from_elem = find_component_by_path(app_state.xml_root_element, from_comp_path) + to_elem = find_component_by_path(app_state.xml_root_element, to_comp_path) + if from_elem is not None and to_elem is not None: + if not live_hooks.live_add_link(from_elem, from_slot_name, to_elem, to_slot_name): + return + link_elem = ET.SubElement(links_elem, "link") link_elem.set("from", from_path_slot) link_elem.set("to", to_path_slot) from_slot = from_path_slot.rsplit(".", 1)[-1] if "." in from_path_slot else from_path_slot to_slot = to_path_slot.rsplit(".", 1)[-1] if "." in to_path_slot else to_path_slot - undo.push(undo.AddLinkCommand( - links_elem, link_elem, - label=f"add link {from_slot}→{to_slot}" - )) + if not app_state.live_mode: + undo.push(undo.AddLinkCommand( + links_elem, link_elem, + label=f"add link {from_slot}→{to_slot}" + )) app_state.mark_dirty() wiresheet.render_current_sheet() @@ -861,6 +904,10 @@ def execute_cascading_deletion(comp_element): if not confirm: return False + if live_hooks.is_live(): + if not live_hooks.live_delete_component(comp_element): + return False + comp_path = get_component_path(comp_element, app_state.xml_root_element) all_elements_to_prune = [comp_element] + descendant_comps @@ -896,7 +943,7 @@ def execute_cascading_deletion(comp_element): comp_index = list(xml_parent).index(comp_element) xml_parent.remove(comp_element) - if xml_parent is not None: + if xml_parent is not None and not app_state.live_mode: undo.push(undo.DeleteComponentCommand( xml_parent, comp_index, comp_element, link_records, label=f"delete {comp_name}" diff --git a/schema.py b/schema.py index 1077a3a..c7e50ed 100644 --- a/schema.py +++ b/schema.py @@ -224,6 +224,18 @@ def find_manifest_path(kit_name, checksum): return None +def find_manifest(kit_name, checksum): + """Returns the parsed ET root of a kit manifest, or None.""" + checksum_hex = f"{checksum:08x}" if isinstance(checksum, int) else checksum + path = find_manifest_path(kit_name, checksum_hex) + if path is None: + return None + try: + return ET.parse(path).getroot() + except Exception: + return None + + def load_schema_kits_and_manifests(): """Loads the manifests for the file's kits, plus any kit still only pending. diff --git a/sox.py b/sox.py new file mode 100644 index 0000000..2fae903 --- /dev/null +++ b/sox.py @@ -0,0 +1,1029 @@ +""" +Sox (Sedona Object Exchange) protocol layer. + +Ported from sedonaOntrol/src/sedona/sox/ Java sources. +Provides component tree access, property read/write, +subscribe/unsubscribe, and structural mutations over DASP. +""" + +import struct +import threading +import time + +from dasp import DaspSocket, DaspException + + +# ──────────────────────────────────────────────────────────────── +# Constants +# ──────────────────────────────────────────────────────────────── + +TREE = 0x01 +CONFIG = 0x02 +RUNTIME = 0x04 +LINKS = 0x08 +ALL = TREE | CONFIG | RUNTIME | LINKS + +# Sedona type ids (matches Type.java constants) +BOOL_ID = 0 +BYTE_ID = 1 +SHORT_ID = 2 +INT_ID = 3 +LONG_ID = 4 +FLOAT_ID = 5 +DOUBLE_ID = 6 +BUF_ID = 7 +STR_ID = 8 + + +class SoxException(Exception): + pass + + +# ──────────────────────────────────────────────────────────────── +# SoxMsg — cursor-based binary buffer for Sox payloads +# ──────────────────────────────────────────────────────────────── + +class SoxMsg: + """Big-endian binary buffer with cursor, mirroring sedona.Buf.""" + + def __init__(self, data=None, capacity=512): + if data is not None: + self._buf = bytearray(data) + self._size = len(data) + else: + self._buf = bytearray(capacity) + self._size = 0 + self._pos = 0 + + # ── Read ──────────────────────────────────────────────────── + + def u1(self): + v = self._buf[self._pos] + self._pos += 1 + return v + + def u2(self): + v = (self._buf[self._pos] << 8) | self._buf[self._pos + 1] + self._pos += 2 + return v + + def i4(self): + v, = struct.unpack_from(">i", self._buf, self._pos) + self._pos += 4 + return v + + def i8(self): + v, = struct.unpack_from(">q", self._buf, self._pos) + self._pos += 8 + return v + + def f4(self): + v, = struct.unpack_from(">f", self._buf, self._pos) + self._pos += 4 + return v + + def f8(self): + v, = struct.unpack_from(">d", self._buf, self._pos) + self._pos += 8 + return v + + def str(self): + start = self._pos + while self._buf[self._pos] != 0: + self._pos += 1 + s = self._buf[start:self._pos].decode("utf-8", errors="replace") + self._pos += 1 # null terminator + return s + + def read_bytes(self, n): + data = bytes(self._buf[self._pos:self._pos + n]) + self._pos += n + return data + + @property + def remaining(self): + return self._size - self._pos + + # ── Write ─────────────────────────────────────────────────── + + def _grow(self, need): + end = self._pos + need + if end > len(self._buf): + new_cap = max(end, len(self._buf) * 2) + self._buf.extend(b'\x00' * (new_cap - len(self._buf))) + if end > self._size: + self._size = end + + def write_u1(self, v): + self._grow(1) + self._buf[self._pos] = v & 0xFF + self._pos += 1 + + def write_u2(self, v): + self._grow(2) + self._buf[self._pos] = (v >> 8) & 0xFF + self._buf[self._pos + 1] = v & 0xFF + self._pos += 2 + + def write_i4(self, v): + self._grow(4) + struct.pack_into(">i", self._buf, self._pos, v) + self._pos += 4 + + def write_i8(self, v): + self._grow(8) + struct.pack_into(">q", self._buf, self._pos, v) + self._pos += 8 + + def write_f4(self, v): + self._grow(4) + struct.pack_into(">f", self._buf, self._pos, v) + self._pos += 4 + + def write_f8(self, v): + self._grow(8) + struct.pack_into(">d", self._buf, self._pos, v) + self._pos += 8 + + def write_str(self, s): + encoded = s.encode("utf-8") + self._grow(len(encoded) + 1) + self._buf[self._pos:self._pos + len(encoded)] = encoded + self._pos += len(encoded) + self._buf[self._pos] = 0 + self._pos += 1 + + def write_bytes(self, data): + self._grow(len(data)) + self._buf[self._pos:self._pos + len(data)] = data + self._pos += len(data) + + # ── Payload ───────────────────────────────────────────────── + + def to_bytes(self): + return bytes(self._buf[:self._size]) + + @property + def command(self): + return self._buf[0] + + @property + def reply_num(self): + return self._buf[1] & 0xFF + + @reply_num.setter + def reply_num(self, v): + self._buf[1] = v & 0xFF + + def check_response(self, expected_cmd): + cmd = self.u1() + _reply = self.u1() + if cmd == ord('!'): + cause = self.str() + raise SoxException(f"Request failed: {cause}") + if cmd != ord(expected_cmd): + raise SoxException( + f"Response {chr(cmd)}({cmd}) != expected {expected_cmd}({ord(expected_cmd)})") + + +# ──────────────────────────────────────────────────────────────── +# Value codec — encode/decode Sedona property values +# ──────────────────────────────────────────────────────────────── + +def decode_value(msg, type_id): + if type_id == BOOL_ID: + return msg.u1() != 0 + elif type_id == BYTE_ID: + return msg.u1() + elif type_id == SHORT_ID: + return msg.u2() + elif type_id == INT_ID: + return msg.i4() + elif type_id == LONG_ID: + return msg.i8() + elif type_id == FLOAT_ID: + return msg.f4() + elif type_id == DOUBLE_ID: + return msg.f8() + elif type_id == STR_ID: + size = msg.u2() + if size == 0: + return "" + chars = msg.read_bytes(size - 1).decode("utf-8", errors="replace") + msg.u1() # null terminator + return chars + elif type_id == BUF_ID: + size = msg.u2() + return msg.read_bytes(size) + else: + raise SoxException(f"Unknown type id: {type_id}") + + +def encode_value(msg, type_id, value): + if type_id == BOOL_ID: + msg.write_u1(1 if value else 0) + elif type_id == BYTE_ID: + msg.write_u1(value & 0xFF) + elif type_id == SHORT_ID: + msg.write_u2(value & 0xFFFF) + elif type_id == INT_ID: + msg.write_i4(value) + elif type_id == LONG_ID: + msg.write_i8(value) + elif type_id == FLOAT_ID: + msg.write_f4(value) + elif type_id == DOUBLE_ID: + msg.write_f8(value) + elif type_id == STR_ID: + encoded = value.encode("utf-8") if isinstance(value, str) else value + msg.write_u2(len(encoded) + 1) + msg.write_bytes(encoded) + msg.write_u1(0) + elif type_id == BUF_ID: + msg.write_u2(len(value)) + msg.write_bytes(value) + else: + raise SoxException(f"Unknown type id: {type_id}") + + +TYPE_ID_BY_NAME = { + "bool": BOOL_ID, "byte": BYTE_ID, "short": SHORT_ID, + "int": INT_ID, "long": LONG_ID, "float": FLOAT_ID, + "double": DOUBLE_ID, "buf": BUF_ID, "Buf": BUF_ID, +} + + +def type_id_for_slot(slot_type_name): + if slot_type_name in TYPE_ID_BY_NAME: + return TYPE_ID_BY_NAME[slot_type_name] + return STR_ID # asStr slots + + +# ──────────────────────────────────────────────────────────────── +# Link +# ──────────────────────────────────────────────────────────────── + +class Link: + __slots__ = ("from_comp_id", "from_slot_id", "to_comp_id", "to_slot_id") + + def __init__(self, from_comp_id=0, from_slot_id=0, to_comp_id=0, to_slot_id=0): + self.from_comp_id = from_comp_id + self.from_slot_id = from_slot_id + self.to_comp_id = to_comp_id + self.to_slot_id = to_slot_id + + def __repr__(self): + return (f"Link({self.from_comp_id}:{self.from_slot_id}" + f" -> {self.to_comp_id}:{self.to_slot_id})") + + +# ──────────────────────────────────────────────────────────────── +# SoxComponent — cached remote component +# ──────────────────────────────────────────────────────────────── + +class SoxComponent: + def __init__(self, comp_id, kit_id=-1, type_id=-1, type_name=""): + self.id = comp_id + self.kit_id = kit_id + self.type_id = type_id + self.type_name = type_name # "kit::Type" qualified name + self.name = "" + self.parent = -1 + self.children = [] # list of child comp ids + self.permissions = 0 + self.subscription = 0 # bitmask of subscribed categories + self.links = [] # list of Link objects + self.values = {} # slot_name -> value (config + runtime) + + def __repr__(self): + return f"SoxComponent(id={self.id}, name={self.name!r}, type={self.type_name})" + + +# ──────────────────────────────────────────────────────────────── +# KitPart — kit name + checksum from schema read +# ──────────────────────────────────────────────────────────────── + +class KitPart: + __slots__ = ("name", "checksum") + + def __init__(self, name, checksum): + self.name = name + self.checksum = checksum + + def __repr__(self): + return f"{self.name}-{self.checksum:08x}" + + +class VersionInfo: + def __init__(self, platform_id, scode_flags, kit_versions, props=None): + self.platform_id = platform_id + self.scode_flags = scode_flags + self.kit_versions = kit_versions # list of (kit_name, checksum, version_str) + self.props = props or {} # extra props like "soxVer" + + +# ──────────────────────────────────────────────────────────────── +# SoxClient — Sox protocol client with component cache +# ──────────────────────────────────────────────────────────────── + +class SoxClient: + """ + Sox client with component cache, request/response correlation, + batch subscribe, and COV event dispatch. + """ + + def __init__(self, host, port, user, password, **dasp_options): + self.host = host + self.port = port + self.user = user + self.password = password + self.dasp_options = dasp_options + + self._dasp_socket = None + self._session = None + self._receiver = None + self._lock = threading.Lock() + + # request/response correlation + self._requests = None # list[SoxMsg] current batch + self._responses = None # list[SoxMsg|None] current batch + self._req_event = threading.Event() + + # cache + self.cache = [None] * 1024 # SoxComponent[] indexed by comp_id + self.all_tree_events = False + + # schema + self.schema_parts = None # list[KitPart] + self.version_info = None # VersionInfo + self.schema = None # will hold schema reference from editor + + # slot registry — set externally after readSchema + # maps (kit_id, type_id) -> list of slot dicts + # each slot dict: {name, id, type_name, is_config, is_runtime, is_action, is_operator, is_as_str} + self.slot_registry = {} + + # event callback: called on main thread via queue + self.on_event = None # fn(comp_id, what_char) + + # ── Connect / Disconnect ──────────────────────────────────── + + def connect(self): + self._dasp_socket = DaspSocket() + self._session = self._dasp_socket.connect( + self.host, self.port, self.user, self.password, + **self.dasp_options) + self._receiver = _SoxReceiver(self) + self._receiver.start() + + def close(self): + if self._receiver: + self._receiver.kill() + self._receiver = None + if self._session and not self._session.is_closed: + try: + self._session.close() + except Exception: + pass + if self._dasp_socket: + try: + self._dasp_socket.close() + except Exception: + pass + self._session = None + self._dasp_socket = None + self.cache = [None] * 1024 + self.all_tree_events = False + + @property + def is_closed(self): + return self._session is None or self._session.is_closed + + # ── Read Schema ───────────────────────────────────────────── + + def read_schema(self): + if self.schema_parts is not None: + return self.schema_parts + req = _prepare_request(ord('v')) + res = self.request(req) + res.check_response('V') + num = res.u1() + parts = [] + for _ in range(num): + name = res.str() + checksum = res.i4() + parts.append(KitPart(name, checksum)) + self.schema_parts = parts + return parts + + def read_version(self): + if self.version_info is not None: + return self.version_info + parts = self.read_schema() + req = _prepare_request(ord('y')) + res = self.request(req) + res.check_response('Y') + platform_id = res.str() + scode_flags = res.u1() + kit_versions = [] + for p in parts: + ver = res.str() + kit_versions.append((p.name, p.checksum, ver)) + vi = VersionInfo(platform_id, scode_flags, kit_versions) + prop_num = res.u1() + for _ in range(prop_num): + k = res.str() + v = res.str() + vi.props[k] = v + self.version_info = vi + return vi + + @property + def sox_version(self): + try: + vi = self.read_version() + return vi.props.get("soxVer") + except Exception: + return None + + # ── Load Component (Tree) ─────────────────────────────────── + + def load_app(self): + return self.load(0) + + def load(self, comp_id): + if comp_id < 0: + return None + cached = self.cache_get(comp_id) + if cached is not None: + return cached + results = self.load_ids([comp_id]) + return results[0] if results else None + + def load_ids(self, ids, checked=True): + self.read_schema() + results = [None] * len(ids) + reqs = [] + req_indices = [] + + for i, cid in enumerate(ids): + c = self.cache_get(cid) + if c is not None: + results[i] = c + else: + reqs.append(_make_update_req(cid, ord('t'))) + req_indices.append(i) + + if not reqs: + return results + + responses = self.requests(reqs) + for res in responses: + res.check_response('C') + self._apply_to_cache(res) + + for i, cid in enumerate(ids): + results[i] = self.cache_get(cid) + if checked and results[i] is None: + raise SoxException(f"Failed to load component {cid}") + + return results + + # ── Update (config / runtime / links) ─────────────────────── + + def update(self, comps, mask): + reqs = [] + for c in comps: + if mask & TREE: + reqs.append(_make_update_req(c.id, ord('t'))) + if mask & CONFIG: + reqs.append(_make_update_req(c.id, ord('c'))) + if mask & RUNTIME: + reqs.append(_make_update_req(c.id, ord('r'))) + if mask & LINKS: + reqs.append(_make_update_req(c.id, ord('l'))) + if not reqs: + return + responses = self.requests(reqs) + for res in responses: + res.check_response('C') + self._apply_to_cache(res) + + # ── Subscribe / Unsubscribe ───────────────────────────────── + + def subscribe(self, comps, mask): + if self.sox_version is not None: + self._batch_subscribe(comps, mask) + else: + self._do_subscribe(comps, mask) + + def _batch_subscribe(self, comps, mask): + to_sub = [c for c in comps if (c.subscription & mask) != mask] + if not to_sub: + return + req = _prepare_request(ord('s')) + req.write_u1(mask) + req.write_u1(len(to_sub)) + for c in to_sub: + req.write_u2(c.id) + res = self.request(req) + res.check_response('S') + for c in to_sub: + c.subscription |= mask + + def _do_subscribe(self, comps, mask): + reqs = [] + for c in comps: + cur = c.subscription + need = mask & ~cur + if need & TREE: + reqs.append(_make_subscribe_req(c.id, ord('t'))) + if need & CONFIG: + reqs.append(_make_subscribe_req(c.id, ord('c'))) + if need & RUNTIME: + reqs.append(_make_subscribe_req(c.id, ord('r'))) + if need & LINKS: + reqs.append(_make_subscribe_req(c.id, ord('l'))) + if not reqs: + return + responses = self.requests(reqs) + for res in responses: + res.check_response('S') + self._apply_to_cache(res) + for c in comps: + c.subscription |= mask + + def subscribe_all_tree_events(self): + if self.sox_version is not None: + req = _prepare_request(ord('s')) + req.write_u1(0xFF) + req.write_u1(0) + self.request(req) + else: + self.request(_make_subscribe_req(0, ord('a'))) + self.all_tree_events = True + for c in self.cache: + if c is not None: + c.subscription |= TREE + + def unsubscribe(self, comps, mask): + if self.is_closed: + return + if self.all_tree_events: + mask &= ~TREE + to_unsub = [c for c in comps if (c.subscription & mask) != 0] + if not to_unsub: + return + if self.sox_version is not None: + self._batch_unsubscribe(to_unsub, mask) + else: + self._do_unsubscribe(to_unsub, mask) + for c in comps: + c.subscription &= ~mask + + def _batch_unsubscribe(self, comps, mask): + req = _prepare_request(ord('u')) + req.write_u1(mask) + req.write_u1(len(comps)) + for c in comps: + req.write_u2(c.id) + self.request(req) + + def _do_unsubscribe(self, comps, mask): + reqs = [] + for c in comps: + reqs.append(_make_unsubscribe_req(c.id, mask)) + self.requests(reqs) + + # ── Read Property ─────────────────────────────────────────── + + def read_prop(self, comp_id, slot_id, type_id): + req = _prepare_request(ord('r')) + req.write_u2(comp_id) + req.write_u1(slot_id) + res = self.request(req) + res.check_response('R') + res.u2() # comp_id + res.u1() # slot_id + res_type_id = res.u1() + return decode_value(res, res_type_id) + + # ── Write Property ────────────────────────────────────────── + + def write_prop(self, comp_id, slot_id, type_id, value): + req = _prepare_request(ord('w')) + req.write_u2(comp_id) + req.write_u1(slot_id) + encode_value(req, type_id, value) + res = self.request(req) + res.check_response('W') + + # ── Invoke Action ─────────────────────────────────────────── + + def invoke(self, comp_id, slot_id, type_id=None, arg=None): + req = _prepare_request(ord('i')) + req.write_u2(comp_id) + req.write_u1(slot_id) + if arg is not None and type_id is not None: + encode_value(req, type_id, arg) + res = self.request(req) + res.check_response('I') + + # ── Add Component ─────────────────────────────────────────── + + def add(self, parent_id, kit_id, type_id_val, name, config_values=None): + req = _prepare_request(ord('a')) + req.write_u2(parent_id) + req.write_u1(kit_id) + req.write_u1(type_id_val) + req.write_str(name) + if config_values: + for tid, val in config_values: + encode_value(req, tid, val) + res = self.request(req) + res.check_response('A') + new_id = res.u2() + + comp = SoxComponent(new_id, kit_id, type_id_val) + comp.name = name + comp.parent = parent_id + comp.children = [] + self.cache_add(comp) + + parent = self.cache_get(parent_id) + if parent is not None and new_id not in parent.children: + parent.children.append(new_id) + + return comp + + # ── Delete Component ──────────────────────────────────────── + + def delete(self, comp_id): + req = _prepare_request(ord('d')) + req.write_u2(comp_id) + res = self.request(req) + res.check_response('D') + + comp = self.cache_get(comp_id) + if comp is not None: + parent = self.cache_get(comp.parent) + if parent is not None and comp_id in parent.children: + parent.children.remove(comp_id) + self.cache_remove(comp) + + # ── Rename Component ──────────────────────────────────────── + + def rename(self, comp_id, new_name): + req = _prepare_request(ord('n')) + req.write_u2(comp_id) + req.write_str(new_name) + res = self.request(req) + res.check_response('N') + comp = self.cache_get(comp_id) + if comp is not None: + comp.name = new_name + + # ── Reorder Children ──────────────────────────────────────── + + def reorder(self, comp_id, child_ids): + req = _prepare_request(ord('o')) + req.write_u2(comp_id) + req.write_u1(len(child_ids)) + for cid in child_ids: + req.write_u2(cid) + res = self.request(req) + res.check_response('O') + comp = self.cache_get(comp_id) + if comp is not None: + comp.children = list(child_ids) + + # ── Link / Unlink ─────────────────────────────────────────── + + def link(self, from_comp_id, from_slot_id, to_comp_id, to_slot_id): + self._link_cmd(ord('a'), from_comp_id, from_slot_id, to_comp_id, to_slot_id) + + def unlink(self, from_comp_id, from_slot_id, to_comp_id, to_slot_id): + self._link_cmd(ord('d'), from_comp_id, from_slot_id, to_comp_id, to_slot_id) + + def _link_cmd(self, subcmd, from_comp_id, from_slot_id, to_comp_id, to_slot_id): + req = _prepare_request(ord('l')) + req.write_u1(subcmd) + req.write_u2(from_comp_id) + req.write_u1(from_slot_id) + req.write_u2(to_comp_id) + req.write_u1(to_slot_id) + res = self.request(req) + res.check_response('L') + + # ── Links snapshot ────────────────────────────────────────── + + def read_links(self, comp_id): + req = _prepare_request(ord('c')) + req.write_u2(comp_id) + req.write_u1(ord('l')) + res = self.request(req) + res.check_response('C') + res_comp_id = res.u2() + what = res.u1() + links = [] + while True: + from_comp = res.u2() + if from_comp == 0xFFFF: + break + from_slot = res.u1() + to_comp = res.u2() + to_slot = res.u1() + links.append(Link(from_comp, from_slot, to_comp, to_slot)) + return links + + # ── Cache Management ──────────────────────────────────────── + + def cache_get(self, comp_id): + if 0 <= comp_id < len(self.cache): + return self.cache[comp_id] + return None + + def cache_add(self, comp): + if comp.id >= len(self.cache): + new_size = max(len(self.cache) * 2, comp.id + 32) + self.cache.extend([None] * (new_size - len(self.cache))) + self.cache[comp.id] = comp + + def cache_remove(self, comp): + for child_id in list(comp.children): + child = self.cache_get(child_id) + if child is not None: + self.cache_remove(child) + if 0 <= comp.id < len(self.cache): + self.cache[comp.id] = None + + # ── Apply to Cache ────────────────────────────────────────── + + def _apply_to_cache(self, msg, is_event=False): + comp_id = msg.u2() + what = msg.u1() + cached = self.cache_get(comp_id) + + if cached is None: + if what == ord('t'): + cached = self._apply_tree_new(msg, comp_id) + self.cache_add(cached) + else: + return + else: + if what == ord('t'): + self._apply_tree_cached(msg, cached) + elif what in (ord('c'), ord('C')): + self._apply_props(msg, cached, 'c', what == ord('C')) + elif what in (ord('r'), ord('R')): + self._apply_props(msg, cached, 'r', what == ord('R')) + elif what == ord('l'): + self._apply_links(msg, cached) + + if is_event and cached is not None: + mask = _what_to_mask(what) + if mask and not (cached.subscription & mask): + cached.subscription |= mask + if self.on_event: + self.on_event(comp_id, chr(what)) + + def _apply_tree_new(self, msg, comp_id): + kit_id = msg.u1() + type_id_val = msg.u1() + type_name = self._resolve_type_name(kit_id, type_id_val) + comp = SoxComponent(comp_id, kit_id, type_id_val, type_name) + self._apply_tree_body(msg, comp) + return comp + + def _apply_tree_cached(self, msg, comp): + _kit_id = msg.u1() + _type_id_val = msg.u1() + self._apply_tree_body(msg, comp) + + def _apply_tree_body(self, msg, comp): + comp.name = msg.str() + comp.parent = msg.u2() + comp.permissions = msg.u1() + num_children = msg.u1() + new_children = [msg.u2() for _ in range(num_children)] + + old_set = set(comp.children) if comp.children else set() + new_set = set(new_children) + for removed_id in old_set - new_set: + removed = self.cache_get(removed_id) + if removed is not None: + self.cache_remove(removed) + + comp.children = new_children + + def _apply_props(self, msg, comp, what_char, operator_only): + slots = self._get_slots(comp.kit_id, comp.type_id) + if slots is None: + return + for slot in slots: + if not slot.get("is_prop", True): + continue + if slot.get("is_action"): + continue + if operator_only and not slot.get("is_operator"): + continue + is_config = slot.get("is_config", True) + is_runtime = slot.get("is_runtime", False) + if what_char == 'c' and is_runtime: + continue + if what_char == 'r' and is_config: + continue + tid = type_id_for_slot(slot.get("type_name", "int")) + try: + val = decode_value(msg, tid) + comp.values[slot["name"]] = val + except Exception: + break + + def _apply_links(self, msg, comp): + links = [] + while msg.remaining >= 6: + from_comp = msg.u2() + if from_comp == 0xFFFF: + break + from_slot = msg.u1() + to_comp = msg.u2() + to_slot = msg.u1() + links.append(Link(from_comp, from_slot, to_comp, to_slot)) + comp.links = links + + def _resolve_type_name(self, kit_id, type_id_val): + if self.schema_parts and 0 <= kit_id < len(self.schema_parts): + kit_name = self.schema_parts[kit_id].name + else: + kit_name = f"kit{kit_id}" + return f"{kit_name}::type{type_id_val}" + + def _get_slots(self, kit_id, type_id_val): + return self.slot_registry.get((kit_id, type_id_val)) + + # ── Networking ────────────────────────────────────────────── + + def request(self, req): + results = self.requests([req]) + return results[0] + + def requests(self, reqs): + if self.is_closed: + raise SoxException("SoxClient closed") + + if len(reqs) > 255: + return self._chunk_requests(reqs) + + with self._lock: + for i, req in enumerate(reqs): + req.reply_num = i + + self._responses = [None] * len(reqs) + self._requests = reqs + self._req_event.clear() + + for req in reqs: + self._session.send(req.to_bytes()) + + deadline = time.monotonic() + 30.0 + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + + if all(r is not None for r in self._responses): + res = list(self._responses) + self._requests = None + self._responses = None + return res + + self._req_event.wait(timeout=min(remaining, 0.5)) + self._req_event.clear() + + if self.is_closed: + raise SoxException("Session closed while waiting for response") + + self._requests = None + self._responses = None + raise SoxException("Request timeout") + + def _chunk_requests(self, reqs): + results = [] + for i in range(0, len(reqs), 255): + chunk = reqs[i:i + 255] + results.extend(self.requests(chunk)) + return results + + # called by receiver thread + def _receive_response(self, msg): + reqs = self._requests + resps = self._responses + if reqs is None or resps is None: + return + + cmd = msg.command + reply_num = msg.reply_num + index = reply_num & 0xFF + + if index < 0 or index >= len(reqs): + return + + req_cmd = reqs[index].command + if cmd != ord('!') and cmd != (req_cmd & ~0x20): + return + + resps[index] = msg + self._req_event.set() + + # called by receiver thread for events + def _receive_event(self, msg): + _cmd = msg.u1() + _reply = msg.u1() + try: + self._apply_to_cache(msg, is_event=True) + except Exception: + pass + + +# ──────────────────────────────────────────────────────────────── +# SoxReceiver — daemon thread +# ──────────────────────────────────────────────────────────────── + +class _SoxReceiver(threading.Thread): + def __init__(self, client): + super().__init__(name="SoxReceiver", daemon=True) + self._client = client + self._alive = True + + def kill(self): + self._alive = False + + def run(self): + client = self._client + while self._alive and not client.is_closed: + try: + dasp_msg = client._session.receive(timeout=1.0) + if dasp_msg is None: + continue + payload = dasp_msg.payload + if not payload or len(payload) < 2: + continue + msg = SoxMsg(payload) + cmd = payload[0] + if cmd == ord('e'): + client._receive_event(msg) + elif cmd == ord('k') or cmd == ord('z'): + pass # file transfer — not needed for editor + else: + client._receive_response(msg) + except DaspException: + if self._alive: + break + except Exception: + if self._alive and not client.is_closed: + continue + + +# ──────────────────────────────────────────────────────────────── +# Message factories +# ──────────────────────────────────────────────────────────────── + +def _prepare_request(cmd, reply_num=0xFF): + msg = SoxMsg() + msg.write_u1(cmd) + msg.write_u1(reply_num) + return msg + + +def _make_update_req(comp_id, what): + req = _prepare_request(ord('c')) + req.write_u2(comp_id) + req.write_u1(what) + return req + + +def _make_subscribe_req(comp_id, what): + req = _prepare_request(ord('s')) + req.write_u2(comp_id) + req.write_u1(what) + return req + + +def _make_unsubscribe_req(comp_id, mask): + req = _prepare_request(ord('u')) + req.write_u2(comp_id) + req.write_u1(mask) + return req + + +def _what_to_mask(what): + if what == ord('t'): + return TREE + if what in (ord('c'), ord('C')): + return CONFIG + if what in (ord('r'), ord('R')): + return RUNTIME + if what == ord('l'): + return LINKS + return 0