wtd-libreflow-addon/scene_manager/ipc.py

"""Client of the Libreflow IPC server (kabaret.ipc) and its wtdcmds commands.

The server writes its address and token in ``config.LIBREFLOW_SERVER_FILE``.
The file is read again on every request: the port changes each time the
server restarts.

``run`` executes a request in a worker thread and hands the result back on
Blender's main thread, so the UI never waits on the network.
"""

import json
import os
import queue
import threading
import traceback
import urllib.error
import urllib.parse
import urllib.request

import bpy

from .. import config

# Hints added to the server's message for the statuses the user can act on.
_STATUS_HINTS = {
    401: "the server refused the token, try again",
    404: "the Libreflow session is gone, or its wtdcmds extension is too old: update it and restart Libreflow",
    504: "no answer within 10 s, the command may still complete in Libreflow",
}
# Statuses whose message replaces the server's.
_STATUS_MESSAGES = {400: "Several Libreflow sessions are connected: pick one in the Scene Manager panel"}


class LibreflowError(Exception):
    """A request failed. ``status`` is the HTTP status, None when the server was not reached."""

    def __init__(self, message, status=None):
        super().__init__(message)
        self.status = status


class NotRunning(LibreflowError):
    """No IPC server: Libreflow is not running."""


def server_info():
    path = os.path.expanduser(config.LIBREFLOW_SERVER_FILE)
    try:
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)
    except FileNotFoundError:
        raise NotRunning("Libreflow is not running") from None
    except (OSError, ValueError) as exc:
        raise LibreflowError(f"Cannot read {path}: {exc}") from None


def _request(method, path, body=None, params=None):
    info = server_info()
    url = f"http://{info['host']}:{info['port']}{path}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    request = urllib.request.Request(
        url,
        data=None if body is None else json.dumps(body).encode(),
        headers={"Authorization": f"Bearer {info['token']}", "Content-Type": "application/json"},
        method=method,
    )
    try:
        with urllib.request.urlopen(request, timeout=config.LIBREFLOW_TIMEOUT) as response:
            return json.loads(response.read() or b"null")
    except urllib.error.HTTPError as exc:
        try:
            detail = json.loads(exc.read() or b"{}").get("detail", exc.reason)
        except ValueError:
            detail = exc.reason
        hint = _STATUS_HINTS.get(exc.code)
        message = _STATUS_MESSAGES.get(exc.code) or (f"{detail} ({hint})" if hint else str(detail))
        raise LibreflowError(message, exc.code) from None
    except (urllib.error.URLError, OSError) as exc:
        raise NotRunning(f"Libreflow is not reachable ({getattr(exc, 'reason', exc)})") from None


def sessions():
    """{session_uid: {session_name, cluster_name, is_gui, current_oid}} of the connected sessions."""
    return _request("GET", "/sessions") or {}


def call(cmd_name, session_uid=None, **kwargs):
    """Run a Flow command and return its result."""
    params = {"session_uid": session_uid} if session_uid else None
    return _request("POST", f"/cmds/Flow/{cmd_name}", {"kwargs": kwargs}, params)["result"]


# --------------------------------------------------------------------------- #
# Background jobs
# --------------------------------------------------------------------------- #
_results = queue.Queue()
_jobs = {}  # job id -> label, for the jobs not finished yet
_generation = 0  # bumped on file load: results of older jobs are dropped
_next_id = 0


def run(label, func, on_success, on_error):
    """Run ``func()`` in a thread, then ``on_success(result)`` or ``on_error(exception)`` on the main thread."""
    global _next_id
    _next_id += 1
    job_id, generation = _next_id, _generation
    _jobs[job_id] = label

    def worker():
        try:
            _results.put((job_id, generation, on_success, func()))
        except Exception as exc:  # noqa: BLE001 - handed to on_error on the main thread
            _results.put((job_id, generation, on_error, exc))

    threading.Thread(target=worker, daemon=True).start()
    if not bpy.app.timers.is_registered(_poll):
        bpy.app.timers.register(_poll, first_interval=0.05)


def busy_label():
    """Label of a running job, or ""."""
    return next(iter(_jobs.values()), "")


def reset():
    """Forget the running jobs (their results will be dropped), e.g. when another file is loaded."""
    global _generation
    _generation += 1
    _jobs.clear()


def _poll():
    while True:
        try:
            job_id, generation, callback, value = _results.get_nowait()
        except queue.Empty:
            break
        if generation != _generation:
            continue
        _jobs.pop(job_id, None)
        try:
            callback(value)
        except Exception:  # noqa: BLE001 - keep the poller alive
            traceback.print_exc()
    for window in bpy.context.window_manager.windows:
        for area in window.screen.areas:
            if area.type == "VIEW_3D":
                area.tag_redraw()
    return 0.1 if _jobs else None