sedona_editor/live_hooks.py

"""
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 base64
import binascii
from tkinter import messagebox

import app_state
from sox import BUF_ID, type_id_for_slot


# Why the last live_* call failed, in words the user can act on. Every failing call sets
# it, so a caller can put it in front of the user with warn_failure().
last_error = ""


def is_live():
    return (app_state.live_mode
            and app_state.device_session is not None
            and app_state.device_session.is_connected)


def _fail(reason):
    global last_error
    last_error = reason
    return None


def _session():
    """The connected DeviceSession, or None with last_error saying why not."""
    if not is_live():
        _fail("The device is not connected.")
        return None
    return app_state.device_session


def warn_failure(title, what, detail=None):
    """Tells the user, in a dialog and the status bar, that `what` did not reach the device.

    `detail` defaults to last_error. Live mode has no undo, so the dialog is the only
    record the user gets that the editor and the device may now disagree.
    """
    reason = detail if detail is not None else last_error
    app_state.set_status(f"{what}: {reason}", warn=True)
    messagebox.showwarning(title, f"{what}.\n\n{reason}", parent=app_state.window)


# ── 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."""
    ds = _session()
    if ds is None:
        return False
    comp_id = _comp_id(element)
    if comp_id is None:
        _fail(f"'{element.get('name', '?')}' has no component id.")
        return False

    slot_info = _find_slot(ds, comp_id, slot_name)
    if slot_info is None:
        return False

    type_id = slot_info["type_id"]
    try:
        coerced = _coerce_value(type_id, value_str)
    except (TypeError, ValueError, binascii.Error) as e:
        _fail(f"'{value_str}' is not a valid value for {slot_name}: {e}")
        return False

    try:
        ds.sox_client.write_prop(comp_id, slot_info["id"], type_id, coerced)
        return True
    except Exception as e:
        _fail(f"The device rejected the write to {slot_name}: {e}")
        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."""
    ds = _session()
    if ds is None:
        return None
    parent_id = _comp_id(parent_element)
    if parent_id is None:
        return _fail(f"The parent '{parent_element.get('name', '?')}' has no component id.")

    kit_id = _kit_id(ds, kit_name)
    if kit_id is None:
        return _fail(f"Kit '{kit_name}' is not installed on the device.")
    type_id_val = _type_id(ds, kit_id, type_name)
    if type_id_val is None:
        return _fail(f"The device's '{kit_name}' kit has no type '{type_name}'.")

    try:
        new_comp = ds.sox_client.add(parent_id, kit_id, type_id_val, comp_name,
                                     list(config_values or []))
        ds.after_add(new_comp.id)
        return new_comp.id
    except Exception as e:
        return _fail(f"The device rejected the add: {e}")


# ── Delete Component ────────────────────────────────────────────

def live_delete_component(element):
    """In live mode, sends a Sox delete to the device. Returns True on success."""
    ds = _session()
    if ds is None:
        return False
    comp_id = _comp_id(element)
    if comp_id is None:
        _fail(f"'{element.get('name', '?')}' has no component id.")
        return False

    descendant_ids = []
    for desc in element.iter():
        if desc.tag == "comp" and desc is not element:
            did = _comp_id(desc)
            if did is not None:
                descendant_ids.append(did)

    try:
        ds.sox_client.delete(comp_id)
        ds.after_delete(comp_id, descendant_ids)
        return True
    except Exception as e:
        _fail(f"The device rejected the delete: {e}")
        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."""
    ds = _session()
    if ds is None:
        return False
    comp_id = _comp_id(element)
    if comp_id is None:
        _fail(f"'{element.get('name', '?')}' has no component id.")
        return False

    try:
        ds.sox_client.rename(comp_id, new_name)
        return True
    except Exception as e:
        _fail(f"The device rejected the rename: {e}")
        return False


# ── Link / Unlink ───────────────────────────────────────────────

def _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name):
    """(from_id, from_slot_id, to_id, to_slot_id), or None with last_error set."""
    if from_element is None or to_element is None:
        return _fail("One end of the link is not a component in the loaded tree.")
    from_id = _comp_id(from_element)
    to_id = _comp_id(to_element)
    if from_id is None or to_id is None:
        return _fail("One end of the link has no component id.")
    from_slot = _find_slot(ds, from_id, from_slot_name)
    if from_slot is None:
        return None
    to_slot = _find_slot(ds, to_id, to_slot_name)
    if to_slot is None:
        return None
    return from_id, from_slot["id"], to_id, to_slot["id"]


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."""
    ds = _session()
    if ds is None:
        return False
    ends = _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name)
    if ends is None:
        return False
    try:
        ds.sox_client.link(*ends)
        return True
    except Exception as e:
        _fail(f"The device rejected the link: {e}")
        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."""
    ds = _session()
    if ds is None:
        return False
    ends = _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name)
    if ends is None:
        return False
    try:
        ds.sox_client.unlink(*ends)
        return True
    except Exception as e:
        _fail(f"The device rejected the unlink: {e}")
        return False


# ── Invoke Action ───────────────────────────────────────────────

def live_invoke(element, slot_name, arg=None):
    """In live mode, sends a Sox invoke. Returns True on success."""
    ds = _session()
    if ds is None:
        return False
    comp_id = _comp_id(element)
    if comp_id is None:
        _fail(f"'{element.get('name', '?')}' has no component id.")
        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:
        _fail(f"The device rejected the invoke of {slot_name}: {e}")
        return False


# ── Helpers ─────────────────────────────────────────────────────

def _comp_id(element):
    if element.tag == "app":
        return 0
    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):
    """{name, id, type_id} for a slot of a cached component, or None with last_error set.

    A Buf goes over the wire as a Buf only when it is binary; an asStr Buf travels as a
    Str, which is how the rest of the editor already reads and writes it.
    """
    comp = ds.sox_client.cache_get(comp_id)
    if comp is None:
        return _fail(f"Component {comp_id} is not loaded from the device.")
    type_name = ds.sox_client.type_name_map.get((comp.kit_id, comp.type_id))
    if type_name and type_name in app_state.final_components_registry:
        for s in app_state.final_components_registry[type_name]:
            if s["name"] == slot_name:
                slot_type = s.get("type") or "int"
                as_str = (s.get("facets") or {}).get("asStr") is True
                return {"name": s["name"], "id": s["slot_id"],
                        "type_id": _wire_type(slot_type, as_str)}
        return _fail(f"{type_name} has no slot '{slot_name}'.")
    slots = ds.sox_client.slot_registry.get((comp.kit_id, comp.type_id))
    if slots is None:
        return _fail(f"No kit manifest describes {type_name or 'this component'}'s slots, "
                     "so the slot cannot be addressed on the device.")
    for s in slots:
        if s["name"] == slot_name:
            return {"name": s["name"], "id": s["id"],
                    "type_id": _wire_type(s.get("type_name") or "int", s.get("is_as_str"))}
    return _fail(f"{type_name or 'This component'} has no slot '{slot_name}'.")


def _wire_type(slot_type, as_str):
    if slot_type in ("sys::Buf", "Buf", "buf") and not as_str:
        return BUF_ID
    return type_id_for_slot(slot_type)


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), full_name in ds.sox_client.type_name_map.items():
        if kid == kit_id and full_name.endswith(f"::{type_name}"):
            return tid
    return None


def _coerce_value(type_id, value_str):
    """The Python value encode_value wants for `value_str` in document form.

    Raises ValueError (or binascii.Error) when the text does not fit the type.
    """
    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, LONG_ID):
        try:
            return int(value_str)
        except ValueError:
            return int(value_str, 0)     # meta and flags can be written as 0x...
    elif type_id in (FLOAT_ID, DOUBLE_ID):
        return float(value_str)
    elif type_id == STR_ID:
        return str(value_str)
    elif type_id == BUF_ID:
        if isinstance(value_str, (bytes, bytearray)):
            return bytes(value_str)
        return base64.b64decode(value_str or "", validate=True)   # the .sax form
    return value_str