sedona_editor/sox.py

"""
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 try_str(self):
        """Like str(), but returns None instead of overrunning a truncated buffer."""
        end = self._buf.find(0, self._pos, self._size)
        if end < 0:
            return None
        return self.str()

    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 & 0xFFFFFFFF: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 = {}

        # fn(kit_name, prefix_bytes) -> signed i4 checksum or None; recovers a
        # checksum the device's reply truncated. Set externally by device_session.
        self.checksum_resolver = None

        # type name map — set externally by device_session
        # maps (kit_id, type_id) -> "kit::TypeName" for proper type resolution
        self.type_name_map = {}

        # event callback: called on main thread via queue
        self.on_event = None        # fn(comp_id, what_char)

        # file transfer: one at a time; its chunks arrive unasked, so the receiver
        # hands them to whichever transfer is open
        self._xfer_lock = threading.Lock()
        self._xfer = None           # _FileGet while a get is open

    # ── 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 = []
        # The device builds this reply in a fixed 512-byte packet and silently
        # drops whatever does not fit, so a device with many kits can cut the
        # list short — sometimes mid-checksum of the last kit.
        for _ in range(num):
            name = res.try_str()
            if name is None:
                break
            if res.remaining >= 4:
                checksum = res.i4()
            else:
                prefix = res.read_bytes(res.remaining)
                checksum = (self.checksum_resolver(name, prefix)
                            if self.checksum_resolver else None)
                if checksum is None:
                    break
            parts.append(KitPart(name, checksum))
        if len(parts) < num:
            raise SoxException(
                f"The device lists {num} kits but its reply was cut off after "
                f"{len(parts)} — the list does not fit in the device's 512-byte "
                f"Sox packet. Remove unused kits from the device's scode, or "
                f"shorten kit names.")
        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')
        # Same 512-byte limit as read_schema: whatever is cut off is left
        # blank rather than failing the connect.
        platform_id = res.try_str() or ""
        # Ontrol's message-length header is for Orion only, as in the Niagara driver.
        # A P-ION on firmware 1.32 takes it on most requests, but a file open sent with
        # it goes unanswered and the device drops off the network for ~20 s.
        if self._session is not None:
            self._session.include_message_len = platform_id.startswith("ontrol-orion")
        scode_flags = res.u1() if res.remaining > 0 else 0
        kit_versions = []
        for p in parts:
            ver = res.try_str()
            kit_versions.append((p.name, p.checksum, ver or ""))
        vi = VersionInfo(platform_id, scode_flags, kit_versions)
        prop_num = res.u1() if res.remaining > 0 else 0
        for _ in range(prop_num):
            k = res.try_str()
            v = res.try_str()
            if k is None or v is None:
                break
            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 = []
        comp.type_name = self._resolve_type_name(kit_id, type_id_val)
        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", False)
            if what_char == 'c' and not is_config:
                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):
        resolved = self.type_name_map.get((kit_id, type_id_val))
        if resolved:
            return resolved
        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))

    # ── File transfer ───────────────────────────────────────────

    def get_file(self, uri, progress=None):
        """Reads a file off the device ('app.sab', 'kits.scode', 'images/x.bin').

        Ported from FileTransfer.getFile: an 'f' request opens the transfer, the
        device then pushes the file as unsolicited 'k' chunks, and a 'z' closes it.
        A file the device does not have comes back as a SoxException from the open.
        progress(done_bytes, total_bytes) is called from this thread as chunks land.
        """
        if self.is_closed:
            raise SoxException("SoxClient closed")
        with self._xfer_lock:
            chunk_size = max(16, self._session.ideal_max - 18)
            req = _prepare_request(ord('f'))
            req.write_str("g")
            req.write_str(uri)
            req.write_i4(0)              # fileSize: unknown on a get
            req.write_u2(chunk_size)
            req.write_u1(0)              # end of headers

            # Set before the open goes out: the first chunks can beat its reply.
            xfer = _FileGet()
            self._xfer = xfer
            opened = closed = False
            try:
                try:
                    res = self.request(req)
                except SoxException as e:
                    # A P-ION on firmware 1.32 never answers a file get -- it drops off
                    # the network for ~20 s, as a restart would -- even for a file it
                    # does not have.
                    raise SoxException(
                        f"The device did not answer the request for {uri} ({e}). Its "
                        f"firmware may not support reading files over Sox; some devices "
                        f"stop responding, or restart, when asked.") from e
                res.check_response('F')
                opened = True
                file_size = res.i4()
                chunk_size = res.u2()
                while res.remaining > 0:     # response headers -- nothing we use
                    if not res.str():
                        break
                    res.str()
                num_chunks = -(-file_size // chunk_size) if chunk_size else 0

                timeout = getattr(self._session, "receive_timeout", 30.0) or 30.0
                while not xfer.done(num_chunks):
                    if self.is_closed:
                        raise SoxException(f"Session closed while reading {uri}")
                    if time.monotonic() - xfer.last_receive > timeout:
                        raise SoxException(f"Timed out reading {uri}")
                    xfer.wait(0.5)
                    if progress:
                        progress(min(file_size, xfer.count() * chunk_size), file_size)

                close = self.request(_prepare_request(ord('z')))
                closed = True
                close.check_response('Z')
            finally:
                self._xfer = None
                # Left open, the transfer keeps the device answering "busy" to the next.
                if opened and not closed and not self.is_closed:
                    try:
                        self.request(_prepare_request(ord('z')))
                    except Exception:
                        pass

            data = bytearray(file_size)
            for num, chunk in xfer.chunks.items():
                if num < num_chunks:
                    start = num * chunk_size
                    data[start:start + len(chunk)] = chunk
            if progress:
                progress(file_size, file_size)
            return bytes(data)

    # ── Networking ──────────────────────────────────────────────

    def request(self, req):
        results = self.requests([req])
        return results[0]

    def requests(self, reqs):
        """Sends the batch and waits for every reply."""
        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


class _FileGet:
    """The chunks of the get in progress, filled by the receiver thread."""

    def __init__(self):
        self.chunks = {}
        self.last_receive = time.monotonic()
        self._event = threading.Event()
        self._lock = threading.Lock()

    def receive_chunk(self, msg):
        msg.u1()                         # 'k'
        msg.u1()                         # replyNum
        num = msg.u2()
        size = msg.u2()
        data = msg.read_bytes(size)
        with self._lock:
            self.chunks[num] = data
            self.last_receive = time.monotonic()
        self._event.set()

    def count(self):
        with self._lock:
            return len(self.chunks)

    def done(self, num_chunks):
        with self._lock:
            return all(n in self.chunks for n in range(num_chunks))

    def wait(self, seconds):
        self._event.wait(seconds)
        self._event.clear()


# ────────────────────────────────────────────────────────────────
# 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'):
                    xfer = client._xfer
                    if xfer is not None:
                        xfer.receive_chunk(msg)
                elif cmd == ord('z'):
                    pass  # close of a put -- the editor only gets files
                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