Python API

Automatically generated from the source docstrings with mkdocstrings. Run mkdocs serve to render this page locally.

Plugin

The OctoPrint plugin implementation: mixins, settings, SimpleAPI and the rule application loop.

PandaBranchPlusPlugin

PandaBranchPlusPlugin()

Bases: SettingsPlugin, AssetPlugin, TemplatePlugin, StartupPlugin, EventHandlerPlugin, SimpleApiPlugin, PrinterCallback

Source code in octoprint_pandabranchplus/__init__.py

def __init__(self):
    super().__init__()
    # Long-lived WebSocket client to the Panda (see panda_ws.py). Created in
    # on_after_startup once the host is configured.
    self._ws = None
    # Rule-engine runtime state (never persisted).
    self._temps = {}  # last known temps: {"bed": .., "tool": .., "chamber": ..}
    self._temp_active = {}  # per-channel hysteresis latch: {(kind, id): bool}
    self._last_switch = {}  # per-channel rate limit: {(kind, id): timestamp}
    self._startup_synced = False  # startup_behaviour applied on first connect
    self._reapply_timer = None  # pending re-run after a rate-limited switch

apply_channel_rules

apply_channel_rules(reason='')

Resolve every channel's target on/off and switch where needed.

Manual channels follow manual_on; auto channels follow the state matrix combined with their temp rule (see rules.py). Only channels whose target differs from the live state get a WS command, rate limited per channel by min_switch_interval.

Source code in octoprint_pandabranchplus/__init__.py

def apply_channel_rules(self, reason=""):
    """Resolve every channel's target on/off and switch where needed.

    Manual channels follow manual_on; auto channels follow the state
    matrix combined with their temp rule (see rules.py). Only channels
    whose target differs from the live state get a WS command, rate
    limited per channel by min_switch_interval.
    """
    if not self._settings.get_boolean(["automation_enabled"]):
        return
    if self._ws is None or not self._ws.connected:
        return

    state = self._current_printer_state()
    combine = self._settings.get(["combine_logic"])
    hysteresis = self._settings.get_int(["temp_hysteresis"])
    min_interval = self._settings.get_int(["min_switch_interval"])
    live = self._ws.current_state()
    self._logger.debug(
        "Applying channel rules (state=%s, reason=%s)", state, reason
    )

    for channel in self._settings.get(["channels"]):
        key = (channel["kind"], int(channel["id"]))
        target, temp_active = rules.resolve_target(
            channel,
            state,
            self._temps,
            self._temp_active.get(key, False),
            combine_logic=combine,
            hysteresis=hysteresis,
        )
        self._temp_active[key] = temp_active
        if target is None:
            continue
        current = live.get(channel["kind"], {}).get(int(channel["id"]))
        want = 1 if target == "on" else 0
        if current == want:
            continue
        now = time.monotonic()
        remaining = min_interval - (now - self._last_switch.get(key, 0))
        if remaining > 0:
            # Rate limited: don't drop the switch, retry once the
            # interval has passed (coalesced into a single timer).
            self._logger.debug(
                "Rate limit: deferring %s %s for %.1fs", key[0], key[1], remaining
            )
            self._schedule_reapply(remaining)
            continue
        try:
            self._ws.set_channel(channel["kind"], channel["id"], want)
            self._last_switch[key] = now
            self._logger.info(
                "Rule engine: %s %s -> %s (state=%s)",
                channel["kind"],
                channel["id"],
                target,
                state,
            )
        except panda_ws.PandaWsError as exc:
            self._logger.warning(
                "Switching %s %s failed: %s", channel["kind"], channel["id"], exc
            )

options: show_root_heading: true members_order: source

WebSocket client

The long-lived connection to the Panda: reconnect with backoff, TCP keepalive, state parsing, error classification.

PandaWsClient

PandaWsClient(logger, host, port=80, path='/ws', on_state=None, on_connection=None, reconnect_min=1, reconnect_max=30, frame_log=False)

Long-lived WebSocket client for the Panda's channel control.

Parameters

logger: A logger instance. host, port, path: Panda address and WebSocket endpoint (default port 80, path /ws). on_state: Callback on_state(channels: dict) invoked with the parsed channel state of every push ({"usb": {id: 0|1}, "mx24v": {id: 0|1}}). on_connection: Callback on_connection(connected: bool) invoked on every connect/disconnect of the socket. reconnect_min, reconnect_max: Backoff window (seconds) between reconnect attempts; the delay doubles from min up to max after each failed attempt and resets on success. frame_log: When true, log every raw frame at DEBUG level (diagnostics).

Source code in octoprint_pandabranchplus/panda_ws.py

def __init__(
    self,
    logger,
    host,
    port=80,
    path="/ws",
    on_state=None,
    on_connection=None,
    reconnect_min=1,
    reconnect_max=30,
    frame_log=False,
):
    self._logger = logger
    self._host = host
    self._port = int(port)
    self._path = path
    self._on_state = on_state
    self._on_connection = on_connection
    self._reconnect_min = max(1, int(reconnect_min))
    self._reconnect_max = max(self._reconnect_min, int(reconnect_max))
    self._frame_log = frame_log

    self._ws = None
    self._thread = None
    self._lock = threading.Lock()
    self._stop_event = threading.Event()
    self._connected = False
    self._delay = self._reconnect_min  # current reconnect backoff
    self._state = {"usb": {}, "mx24v": {}}  # last known channel states

current_state

current_state()

Return the last known channel state snapshot.

Source code in octoprint_pandabranchplus/panda_ws.py

def current_state(self):
    """Return the last known channel state snapshot."""
    with self._lock:
        return {kind: dict(states) for kind, states in self._state.items()}

set_channel

set_channel(kind, channel_id, on)

Switch a channel on/off.

Sends {"<kind>": {"id": id, "on": 0|1}}. kind is "usb" or "mx24v". Raises :class:PandaWsError on send failure. The Panda confirms by broadcasting the new state, which arrives via on_state.

Source code in octoprint_pandabranchplus/panda_ws.py

def set_channel(self, kind, channel_id, on):
    """Switch a channel on/off.

    Sends ``{"<kind>": {"id": id, "on": 0|1}}``. ``kind`` is ``"usb"`` or
    ``"mx24v"``. Raises :class:`PandaWsError` on send failure. The Panda
    confirms by broadcasting the new state, which arrives via ``on_state``.
    """
    if kind not in ("usb", "mx24v"):
        raise ValueError(f"unknown channel kind: {kind!r}")
    payload = json.dumps({kind: {"id": int(channel_id), "on": 1 if on else 0}})
    ws = self._ws
    if ws is None or not self._connected:
        raise PandaWsError("send_failed", "not connected to the Panda")
    try:
        ws.send(payload)
    except Exception as exc:
        raise PandaWsError("send_failed", str(exc)) from exc

start

start()

Open the connection and start the background receive/reconnect loop.

Source code in octoprint_pandabranchplus/panda_ws.py

def start(self):
    """Open the connection and start the background receive/reconnect loop."""
    with self._lock:
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._run_loop, name="PandaWsClient", daemon=True
        )
        self._thread.start()

stop

stop()

Close the connection and stop the loop. Idempotent.

Source code in octoprint_pandabranchplus/panda_ws.py

def stop(self):
    """Close the connection and stop the loop. Idempotent."""
    self._stop_event.set()
    ws = self._ws
    if ws is not None:
        with suppress(Exception):
            ws.close()
    thread = self._thread
    if thread is not None and thread.is_alive():
        thread.join(timeout=5)
    self._thread = None

options: show_root_heading: true members_order: source

PandaWsError

PandaWsError(reason, message='')

Bases: Exception

A Panda WebSocket operation failed, carrying a classified reason.

reason is one of unreachable / timeout / send_failed so the UI can translate it without parsing raw socket errors.

Source code in octoprint_pandabranchplus/panda_ws.py

def __init__(self, reason, message=""):
    super().__init__(message or reason)
    self.reason = reason

options: show_root_heading: true

test_connection

test_connection(host, port=80, path='/ws', timeout=5)

One-shot connect: open the socket, read the snapshot, close.

Returns {"channels": {...}, "raw_keys": [...]} on success. Raises :class:PandaWsError with a classified reason on failure. Used by the settings dialog's "Test connection" button — runs against the entered values, independent of the long-lived client.

Source code in octoprint_pandabranchplus/panda_ws.py

def test_connection(host, port=80, path="/ws", timeout=5):
    """One-shot connect: open the socket, read the snapshot, close.

    Returns ``{"channels": {...}, "raw_keys": [...]}`` on success. Raises
    :class:`PandaWsError` with a classified reason on failure. Used by the
    settings dialog's "Test connection" button — runs against the entered
    values, independent of the long-lived client.
    """
    url = f"ws://{host}:{int(port)}{path}"
    try:
        ws = websocket.create_connection(url, timeout=timeout)
    except (websocket.WebSocketTimeoutException, socket.timeout) as exc:
        raise PandaWsError("timeout", str(exc)) from exc
    except (OSError, websocket.WebSocketException) as exc:
        raise PandaWsError("unreachable", str(exc)) from exc

    try:
        try:
            frame = ws.recv()
        except (websocket.WebSocketTimeoutException, socket.timeout) as exc:
            raise PandaWsError("timeout", f"connected, but no snapshot: {exc}") from exc
        except (OSError, websocket.WebSocketException) as exc:
            raise PandaWsError("unreachable", str(exc)) from exc
        try:
            data = json.loads(frame)
        except ValueError as exc:
            raise PandaWsError(
                "unreachable", "connected, but got a non-JSON frame"
            ) from exc
        channels = _parse_channel_state(data)
        if channels is None:
            raise PandaWsError(
                "unreachable",
                "connected, but the snapshot has no usb/mx24v channels"
                " — is this really a Panda Branch Plus?",
            )
        return {"channels": channels, "raw_keys": sorted(data.keys())}
    finally:
        with suppress(Exception):
            ws.close()

options: show_root_heading: true

Rule engine

Pure functions — no OctoPrint imports, fully unit-tested.

rules

Pure rule engine for the channel automation.

Kept free of OctoPrint imports so the decision logic is unit-testable in isolation. The plugin resolves the current printer state and temperatures, then asks :func:resolve_target what each channel should do.

Semantics (see .ideas/channel-automation-plan.md):

• mode: "manual" always wins: the channel follows manual_on, automation never touches it.
• mode: "auto": the state matrix rules[state] decides — "on", "off" or "ignore" (= leave the channel exactly as it is).
• An enabled temp_rule fires when the sensor crosses threshold (with hysteresis on the falling edge) and contributes above ("on"/"off") while active; while inactive it has no opinion.
• combine_logic merges the two verdicts:
• temp_override: an active temp rule replaces the state verdict.
• and: on only if both say on, otherwise off.
• or: on if either says on, otherwise off.

evaluate_temp_rule

evaluate_temp_rule(rule, temps, was_active, hysteresis)

Evaluate a channel's temperature rule.

Parameters

rule: The channel's temp_rule dict (enabled, sensor, threshold, above). temps: Last known temperatures, e.g. {"bed": 62.1, "tool": 210.4}. Missing sensors simply keep the previous verdict (no flapping on gaps in the data). was_active: Whether the rule was active after the previous evaluation (needed for the hysteresis dead band). hysteresis: Dead band in degrees applied on the falling edge.

Returns (active, target) where target is rule["above"] while active and None (no opinion) while inactive.

Source code in octoprint_pandabranchplus/rules.py

def evaluate_temp_rule(rule, temps, was_active, hysteresis):
    """Evaluate a channel's temperature rule.

    Parameters
    ----------
    rule:
        The channel's ``temp_rule`` dict (``enabled``, ``sensor``,
        ``threshold``, ``above``).
    temps:
        Last known temperatures, e.g. ``{"bed": 62.1, "tool": 210.4}``.
        Missing sensors simply keep the previous verdict (no flapping on
        gaps in the data).
    was_active:
        Whether the rule was active after the previous evaluation (needed
        for the hysteresis dead band).
    hysteresis:
        Dead band in degrees applied on the falling edge.

    Returns ``(active, target)`` where ``target`` is ``rule["above"]`` while
    active and ``None`` (no opinion) while inactive.
    """
    if not rule or not rule.get("enabled"):
        return False, None

    action = rule.get("above") or "on"
    temp = temps.get(rule.get("sensor") or "bed")
    if temp is None:
        # No reading for this sensor -> keep the previous verdict.
        return was_active, (action if was_active else None)

    threshold = float(rule.get("threshold") or 0)
    if was_active:
        active = temp > threshold - float(hysteresis)
    else:
        active = temp >= threshold
    return active, (action if active else None)

map_octoprint_state

map_octoprint_state(state_id)

Map an OctoPrint state id to the plugin's five canonical states.

STARTING is what BambuConnector reports for Bambu's PREPARE phase. Anything unknown (offline, detecting, ...) counts as idle — no printer means no print is running (see the plan's state resolution).

Source code in octoprint_pandabranchplus/rules.py

def map_octoprint_state(state_id):
    """Map an OctoPrint state id to the plugin's five canonical states.

    ``STARTING`` is what BambuConnector reports for Bambu's PREPARE phase.
    Anything unknown (offline, detecting, ...) counts as ``idle`` — no
    printer means no print is running (see the plan's state resolution).
    """
    mapping = {
        "STARTING": "prepare",
        "PRINTING": "printing",
        "RESUMING": "printing",
        "FINISHING": "printing",
        "PAUSED": "paused",
        "PAUSING": "paused",
        "ERROR": "error",
        "CLOSED_WITH_ERROR": "error",
    }
    return mapping.get(state_id, "idle")

resolve_target

resolve_target(channel, state, temps, was_active, combine_logic='temp_override', hysteresis=2)

Resolve what a channel should do right now.

Parameters

channel: The channel's stored config (mode, manual_on, rules, temp_rule). state: Canonical printer state: idle/prepare/printing/ paused/error. temps / was_active / hysteresis: See :func:evaluate_temp_rule. combine_logic: temp_override | and | or.

Returns (target, temp_active) where target is "on", "off" or None (= leave the channel untouched).

Source code in octoprint_pandabranchplus/rules.py

def resolve_target(
    channel, state, temps, was_active, combine_logic="temp_override", hysteresis=2
):
    """Resolve what a channel should do right now.

    Parameters
    ----------
    channel:
        The channel's stored config (``mode``, ``manual_on``, ``rules``,
        ``temp_rule``).
    state:
        Canonical printer state: ``idle``/``prepare``/``printing``/
        ``paused``/``error``.
    temps / was_active / hysteresis:
        See :func:`evaluate_temp_rule`.
    combine_logic:
        ``temp_override`` | ``and`` | ``or``.

    Returns ``(target, temp_active)`` where ``target`` is ``"on"``, ``"off"``
    or ``None`` (= leave the channel untouched).
    """
    if channel.get("mode") == "manual":
        return ("on" if channel.get("manual_on") else "off"), was_active

    state_rule = (channel.get("rules") or {}).get(state, "off")
    state_target = None if state_rule == "ignore" else state_rule

    temp_active, temp_target = evaluate_temp_rule(
        channel.get("temp_rule") or {}, temps, was_active, hysteresis
    )

    if temp_target is None:
        return state_target, temp_active

    if combine_logic == "and":
        target = "on" if (state_target == "on" and temp_target == "on") else "off"
    elif combine_logic == "or":
        target = "on" if (state_target == "on" or temp_target == "on") else "off"
    else:  # temp_override (default)
        target = temp_target
    return target, temp_active

options: show_root_heading: true members_order: source

Hardware layout

hardware

Fixed hardware layout of the Panda Branch Plus.

The Panda has 10 switchable outputs the firmware exposes under two WebSocket roots: usb (5x Type-C) and mx24v (5x MX3.0 24V). The channel type is fixed by the hardware and is shown as a badge in the UI; only the label is user-editable. Verified layout: see .ideas/panda-branch-plus-recon.md.

channel_type

channel_type(kind, channel_id)

Return the fixed type string for a channel, or None if unknown.

Source code in octoprint_pandabranchplus/hardware.py

def channel_type(kind, channel_id):
    """Return the fixed type string for a channel, or ``None`` if unknown."""
    return CHANNEL_TYPES.get((kind, channel_id))

options: show_root_heading: true members_order: source