diff --git a/kobrax_client.py b/kobrax_client.py index 320ecae..d3c7e44 100644 --- a/kobrax_client.py +++ b/kobrax_client.py @@ -33,6 +33,7 @@ import sys import threading import time import uuid +from collections import deque from datetime import datetime import env_loader @@ -125,7 +126,7 @@ class KobraXClient: # Pending requests by msgid (for response ACK) self._pending_msgid: dict[str, dict] = {} # Pending requests by msg_type/report topic suffix - self._pending_report: dict[str, dict] = {} + self._pending_report: dict[str, list[dict]] = {} # Optional callbacks: topic_suffix → callable(payload_dict) self.callbacks: dict[str, callable] = {} @@ -321,6 +322,17 @@ class KobraXClient: "target_nozzle_temp", "target_hotbed_temp"}} return hashlib.md5(json.dumps(stable, sort_keys=True).encode(), usedforsecurity=False).hexdigest() + def _is_set_dry_placeholder_ack(self, msg_type: str, action: str, payload: dict | None) -> bool: + if not isinstance(payload, dict): + return False + if msg_type != "multiColorBox" or action != "setDry": + return False + try: + code = int(payload.get("code", -1)) + except Exception: + code = -1 + return code == 0 and not payload.get("action") and payload.get("data") is None + def _dispatch(self, topic: str, payload: dict): suffix = "/".join(topic.split("/")[-2:]) @@ -348,16 +360,18 @@ class KobraXClient: # Resolve by report topic suffix (e.g. "info/report") if suffix in self._pending_report: - entry = self._pending_report[suffix] - entry["result"] = payload - entry["event"].set() + for entry in list(self._pending_report[suffix]): + with entry["lock"]: + entry["results"].append(payload) + entry["event"].set() # Resolve by msgid (for generic response ACK) msgid = payload.get("msgid") if msgid and msgid in self._pending_msgid: entry = self._pending_msgid[msgid] - entry["result"] = payload - entry["event"].set() + with entry["lock"]: + entry["results"].append(payload) + entry["event"].set() # User callbacks by topic suffix (last two path components) if suffix in self.callbacks: @@ -393,13 +407,10 @@ class KobraXClient: # Also register by report topic as fallback for responses without msgid. report_key = f"{msg_type}/report" event = threading.Event() - entry = {"event": event, "result": None} + entry = {"event": event, "results": deque(), "lock": threading.Lock()} self._pending_msgid[msgid] = entry - # Only register report-key waiter if nobody else is waiting on it - report_registered = False - if report_key not in self._pending_report: - self._pending_report[report_key] = entry - report_registered = True + report_waiters = self._pending_report.setdefault(report_key, []) + report_waiters.append(entry) topic = self._pub_topic(msg_type) # Status-Poll-TX (query/getInfo) ist reines Rauschen (alle paar Sekunden) → @@ -414,8 +425,10 @@ class KobraXClient: except Exception as e: log.error("send error: %s, reconnecting…", e) self._pending_msgid.pop(msgid, None) - if report_registered: - self._pending_report.pop(report_key, None) + if entry in report_waiters: + report_waiters.remove(entry) + if not report_waiters: + self._pending_report.pop(report_key, None) if not self._reconnect(): return None # retry once after reconnect @@ -423,24 +436,46 @@ class KobraXClient: with self._lock: self._sock.sendall(_build_publish(topic, payload)) self._pending_msgid[msgid] = entry - if report_registered: - self._pending_report[report_key] = entry except Exception: return None if timeout <= 0: self._pending_msgid.pop(msgid, None) - if report_registered: - self._pending_report.pop(report_key, None) + if entry in report_waiters: + report_waiters.remove(entry) + if not report_waiters: + self._pending_report.pop(report_key, None) return None - received = event.wait(timeout) + deadline = time.monotonic() + timeout + result = None + while True: + candidate = None + with entry["lock"]: + if entry["results"]: + candidate = entry["results"].popleft() + if not entry["results"]: + event.clear() + if candidate is not None: + if self._is_set_dry_placeholder_ack(msg_type, action, candidate): + continue + result = candidate + break + + remaining = deadline - time.monotonic() + if remaining <= 0: + break + if not event.wait(remaining): + break + self._pending_msgid.pop(msgid, None) - if report_registered: - self._pending_report.pop(report_key, None) - if not received: + if entry in report_waiters: + report_waiters.remove(entry) + if not report_waiters: + self._pending_report.pop(report_key, None) + if result is None: return None - return entry["result"] + return result def publish_web(self, msg_type: str, action: str, data=None) -> None: """Fire-and-forget publish on the web/printer topic (used for runtime updates during print).""" diff --git a/kobrax_moonraker_bridge.py b/kobrax_moonraker_bridge.py index 37b8f29..ffde588 100644 --- a/kobrax_moonraker_bridge.py +++ b/kobrax_moonraker_bridge.py @@ -3085,10 +3085,19 @@ class KobraXBridge: loop = asyncio.get_event_loop() def _send(): - return self.client.publish("multiColorBox", "setDry", payload, timeout=0) - # Fire-and-forget: setDry ACK arrives via multiColorBox/report callback. - # Waiting for a response on that busy push topic causes false "code:0" rejections. - await loop.run_in_executor(None, _send) + return self.client.publish("multiColorBox", "setDry", payload, timeout=5) + + resp = await loop.run_in_executor(None, _send) + if resp is None: + return web.json_response({"error": "No response from printer"}, status=504) + + try: + code = int(resp.get("code", -1)) + except Exception: + code = -1 + action_resp = str(resp.get("action", "") or "") + if action_resp != "setDry" or code != 200: + return web.json_response({"error": f"Unexpected dryer response: {resp}"}, status=502) self._state["ace_drying"] = ui_state self._state_dirty = True