From e7d377760ea89a72fdd4fc880c78c9847fa942ea Mon Sep 17 00:00:00 2001 From: Henrik Lemke Date: Mon, 2 Nov 2026 14:52:05 +0100 Subject: [PATCH] eco web: task queues -- what runs and waits, scan progress, output, controls - queues page: per queue the running task with its scan's step n/N, the waiting and the last finished tasks (errors), each task's output live - controls with writes on (all items, since a queue moves whatever its tasks move): pause/resume the queue and the running scan, stop the running task, move a waiting task up/to the top, remove it, drop all waiting, stop everything -- nothing adds or repeats a task - task lines name devices instead of their repr (a whole status table) - demo: a paused queue with three simulated scans of stage.x --- docs/web_app.md | 6 ++ eco/web/__main__.py | 7 +- eco/web/backends.py | 11 ++- eco/web/client_session.py | 37 ++++++++ eco/web/demo.py | 65 +++++++++++++ eco/web/queues.py | 152 +++++++++++++++++++++++++++++++ eco/web/static/app.js | 149 +++++++++++++++++++++++++++++- eco/web/static/style.css | 26 ++++++ eco/web/static/sw.js | 2 +- tests/test_web_client_session.py | 99 ++++++++++++++++++++ web_frontend/PARITY.md | 4 +- web_frontend/PROTOCOL.md | 6 +- web_frontend/README.md | 5 +- 13 files changed, 559 insertions(+), 10 deletions(-) create mode 100644 eco/web/queues.py diff --git a/docs/web_app.md b/docs/web_app.md index 54ccec0..8a65c82 100644 --- a/docs/web_app.md +++ b/docs/web_app.md @@ -76,6 +76,12 @@ python -m eco.web --demo --writes laptop), or use Zoom +/−, Wide and Home. *Settings* shows pan, tilt, zoom, iris and focus as rows. All viewers share one connection to the camera; the stream pauses while the page is in the background. +- **Queues** (start page): the session's task queues ({doc}`queues`): the + running task with the step its scan is at, the waiting tasks, the last + finished ones with their errors, and each task's output, live. With writes + on: pause/resume the queue, pause/resume the running scan at its next step, + stop the running task, move a waiting task up or remove it, drop all + waiting, stop everything. Nothing can be added to a queue from here. - **Panel**: a device's schematic, refreshed every 3 s. Tapping a symbol opens that device; the panel's commands are not run from the web app. diff --git a/eco/web/__main__.py b/eco/web/__main__.py index 3ac0785..57f57d4 100644 --- a/eco/web/__main__.py +++ b/eco/web/__main__.py @@ -51,10 +51,13 @@ def main(argv=None): if args.demo: from eco.web.demo import DemoNamespace + from eco.web.demo import demo_queues + namespace = DemoNamespace() + backend = SessionBackend(namespace, queues=demo_queues(namespace)) else: - namespace = scope_namespace(args.scope, args.lazy) - gateway = WebGateway(SessionBackend(namespace), host=args.host, port=args.port, + backend = SessionBackend(scope_namespace(args.scope, args.lazy)) + gateway = WebGateway(backend, host=args.host, port=args.port, token=args.token, writes=args.writes).start() print(connection_help(gateway)) # a terminated server must still stop the jogs its clients hold diff --git a/eco/web/backends.py b/eco/web/backends.py index efa805b..1b4c294 100644 --- a/eco/web/backends.py +++ b/eco/web/backends.py @@ -40,8 +40,9 @@ def attach_ca_context(): class SessionBackend: - def __init__(self, namespace, label=None): + def __init__(self, namespace, label=None, queues=None): self.namespace = namespace + self._queues = queues # default: the session's eco.utilities.task_queue.queues self.label = label or getattr(namespace, "name", None) or "eco" self._init_lock = threading.Lock() self._index_lock = threading.Lock() @@ -98,6 +99,14 @@ class SessionBackend: obj = obj.__wrapped__ return obj + def queues(self): + """{name: TaskQueue} of the session (eco.utilities.task_queue.queues).""" + if self._queues is not None: + return self._queues + from eco.utilities import task_queue + + return task_queue.queues + def search_index(self): """[(full name, kind, object)] of every adjustable/detector in the initialized components, rebuilt when that set changes. Shared by all diff --git a/eco/web/client_session.py b/eco/web/client_session.py index d2f60b3..3599d78 100644 --- a/eco/web/client_session.py +++ b/eco/web/client_session.py @@ -447,3 +447,40 @@ class ClientSession: raise ProtocolError(f"unknown ptz action {action!r}") logger.info("web client %s: %s ptz %s", self.client_id, name, action) return {"done": action} + + # -- task queues: what runs and waits, and the controls for it ----------------------------- + + def _queue(self, name): + queue = dict(self.backend.queues()).get(name) # a copy: queues[missing] would create one + if queue is None: + raise NotFound(f"no queue {name!r}") + return queue + + def _op_queues(self): + """Every queue of the session (eco.web.queues.queue_json).""" + from eco.web import queues + + return {"queues": [queues.queue_json(q) for q in list(dict(self.backend.queues()).values())], + "now": time.time()} + + def _op_queue_output(self, queue, task, since=0): + """What task #`task` of `queue` printed after character `since`.""" + from eco.web import queues + + found = queues.find_task(self._queue(queue), task) + if found is None: + raise NotFound(f"no task #{task} in queue {queue!r}") + text, offset, clipped = queues.output_since(found, since) + return {"text": text, "offset": offset, "clipped": clipped, "task": queues.task_json(found)} + + def _op_queue_action(self, queue, action, task=None): + """Pause/resume/stop/clear a queue, its running scan, or move/remove a + waiting task (eco.web.queues.ACTIONS) -- writes must be on. Nothing + here adds or starts a new task.""" + from eco.web import queues + + if self.writes is not True: # not per item: a queue moves whatever its tasks move + raise WritesDisabled("queue controls need writes enabled for this web session") + result = queues.act(self._queue(queue), action, task) + logger.info("web client %s: queue %s %s %s", self.client_id, queue, action, task or "") + return {"done": result} diff --git a/eco/web/demo.py b/eco/web/demo.py index 51438d6..5920be5 100644 --- a/eco/web/demo.py +++ b/eco/web/demo.py @@ -373,6 +373,71 @@ class Sample(Assembly): self._append(Cryostat, name="cryo", is_display="recursive") +class SimScan: + """The parts of eco.acquisition.scan.StepScan the queue view reads: + steps done/to do, pause/resume at the next step, stop.""" + + def __init__(self, positions): + self._values_todo = list(positions) + self._values_done = [] + self._pause_event = threading.Event() + self._pause_event.set() + self._stop_requested = False + + def pause(self): + self._pause_event.clear() + + def resume(self): + self._pause_event.set() + + def is_paused(self): + return not self._pause_event.is_set() + + def request_stop(self): + self._stop_requested = True + self._pause_event.set() + + +def demo_scan(motor, detector, start, stop, steps, step_time=0.5): + """A queued 'scan' of a simulated motor: registers a SimScan with its + task (as StepScan does), moves step by step, prints a line per step.""" + from eco.utilities import task_queue + + positions = [start + (stop - start) * i / (steps - 1) for i in range(steps)] + scan = SimScan(positions) + task = task_queue.current_task() + if task is not None: + task.scans.append(scan) + print(f"scan {motor.name} {start} .. {stop}, {steps} steps") + while scan._values_todo: + while not scan._pause_event.wait(0.1): + pass + if scan._stop_requested: + print("stopped on request") + return + value = scan._values_todo[0] + motor.set_target_value(value).wait() + time.sleep(step_time) + print(f"step {len(scan._values_done) + 1:3d}/{steps} {motor.name} = {motor.get_current_value():8.4f}" + f" {detector.name} = {detector.get_current_value():9.2f}") + scan._values_done.append(scan._values_todo.pop(0)) + print("scan done") + + +def demo_queues(namespace): + """A queue "default" with three scans waiting, paused (so they wait for + Resume in the web client), and an empty queue "align".""" + from eco.utilities.task_queue import TaskQueue + + stage = namespace.resolve_item("stage") + default = TaskQueue("default", namespace={}) + default.pause() # before the first append starts its worker + for start, stop, steps in ((-1, 1, 40), (-0.5, 0.5, 30), (0, 2, 20)): + default.append(demo_scan, stage.x, stage.diode, start, stop, steps, test=False, + task_description=f"x {start}..{stop}") + return {"default": default, "align": TaskQueue("align", namespace={})} + + def _broken(name=None): raise ConnectionError("simulated: IOC not reachable") diff --git a/eco/web/queues.py b/eco/web/queues.py new file mode 100644 index 0000000..74087cb --- /dev/null +++ b/eco/web/queues.py @@ -0,0 +1,152 @@ +"""The session's task queues (eco.utilities.task_queue) for the web client: +what runs, what waits, how far the running scan is, what a task printed -- +and the controls of the Qt queue panel (ScanQueuePanel) that act on what is +already queued. Nothing here adds, repeats or edits a task. +""" +import re + +from eco.web.viewmodels import _clip + +HISTORY = 10 # finished tasks listed per queue +OUTPUT_CHUNK = 20000 # characters of output per reply at most (the newest) + +# action -> (needs a task id, what it calls) +ACTIONS = { + "pause": False, # don't start the next task + "resume": False, + "clear": False, # drop every waiting task + "stop": False, # stop the running task and drop the waiting ones + "pause_running": False, # pause the running scan at its next step + "resume_running": False, + "stop_running": False, # its scans stop at their next step, else KeyboardInterrupt + "remove": True, # drop one waiting task + "move_up": True, + "move_top": True, +} + + +def _scan(task): + scan = task.scan + if scan is None: + return None + try: + done = len(getattr(scan, "_values_done", [])) + total = done + len(getattr(scan, "_values_todo", [])) + paused = bool(scan.is_paused()) + except Exception: + return None + return {"done": done, "total": total, "paused": paused, + "stopping": bool(getattr(scan, "_stop_requested", False))} + + +_ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + + +def _short(value): + """An argument as the task line shows it: an eco item by its name (its + repr is a whole status table), plain values by repr, anything else by + its type.""" + alias = getattr(value, "alias", None) + if alias is not None: + try: + return alias.get_full_name() + except Exception: + pass + if value is None or isinstance(value, (bool, int, float, complex, str, bytes)): + return repr(value) + if isinstance(value, (list, tuple)) and len(value) <= 8: + inner = ", ".join(_short(v) for v in value) + return f"[{inner}]" if isinstance(value, list) else f"({inner}{',' if len(value) == 1 else ''})" + if isinstance(value, dict) and len(value) <= 8: + return "{" + ", ".join(f"{k!r}: {_short(v)}" for k, v in value.items()) + "}" + return f"<{type(value).__name__}>" + + +def task_source(task): + """Task.source, but with short arguments (see _short).""" + if isinstance(task.command, str): + return _ANSI.sub("", task.command.strip()) + name = getattr(task.command, "__qualname__", None) or type(task.command).__name__ + parts = [_short(a) for a in task.args] + [f"{k}={_short(v)}" for k, v in task.kwargs.items()] + return f"{name}({', '.join(parts)})" + + +def task_json(task): + out = { + "id": task.id, + "source": _clip(task_source(task)), + "description": _clip(str(task.description or "")), + "state": task.state, + "started": task.started, + "finished": task.finished, + } + if task.exception is not None: + out["error"] = _clip(f"{type(task.exception).__name__}: {task.exception}") + scan = _scan(task) if task.state == "running" else None + if scan is not None: + out["scan"] = scan + return out + + +def queue_json(queue): + current = queue.current + return { + "name": queue.name, + "paused": queue.paused, + "running": task_json(current) if current is not None else None, + "pending": [task_json(t) for t in queue.pending], + "history": [task_json(t) for t in reversed(queue.history[-HISTORY:])], + } + + +def find_task(queue, task_id): + return queue._find(int(task_id), everywhere=True) + + +def output_since(task, since=0): + """(text, offset): what `task` printed after character `since`, at most + OUTPUT_CHUNK characters (the newest). The oldest output of a very chatty + task is dropped by the queue itself; an offset beyond the end starts over.""" + text = task.output + since = int(since or 0) + if since > len(text): + since = 0 + chunk = text[since:] + clipped = len(chunk) > OUTPUT_CHUNK + if clipped: + chunk = chunk[-OUTPUT_CHUNK:] + return chunk, len(text), clipped + + +def act(queue, action, task_id=None): + """Run one allowed action on `queue`; returns a short description.""" + if action not in ACTIONS: + raise ValueError(f"unknown queue action {action!r}") + if ACTIONS[action]: + if task_id is None: + raise ValueError(f"{action} needs a task") + task = queue._find(int(task_id)) + if task is None: + raise ValueError(f"task #{task_id} is not waiting in queue {queue.name!r} (any more)") + if action == "remove": + queue.remove(task) + else: + pending = queue.pending + index = 0 if action == "move_top" else max(0, pending.index(task) - 1) + queue.move(task, index) + return f"task #{task.id}: {action}" + if action == "pause": + queue.pause() + elif action == "resume": + queue.resume() + elif action == "clear": + return f"dropped {len(queue.clear())} waiting tasks" + elif action == "stop": + dropped = queue.stop() + return f"stopping the running task, dropped {len(dropped)} waiting" + elif action in ("pause_running", "resume_running", "stop_running"): + method = {"pause_running": queue.pause_running_task, "resume_running": queue.resume_running_task, + "stop_running": queue.stop_running_task}[action] + if method() is None: + raise ValueError("nothing is running" if action == "stop_running" else "the running task has no scan to pause") + return action diff --git a/eco/web/static/app.js b/eco/web/static/app.js index b23519d..920aa98 100644 --- a/eco/web/static/app.js +++ b/eco/web/static/app.js @@ -293,6 +293,8 @@ function route() { else if (hash.startsWith("/tweak/")) showTweak(hash.slice(7)); else if (hash.startsWith("/jog/")) showJog(hash.slice(5)); else if (hash.startsWith("/cam/")) showCamera(hash.slice(5)); + else if (hash === "/queues") showQueues(); + else if (hash.startsWith("/qout/")) showTaskOutput(hash.slice(6)); else showHome(); } @@ -342,7 +344,9 @@ async function showHome() { autocapitalize: "off", autocomplete: "off", spellcheck: "false", oninput: (e) => { S.filter = e.target.value; render(); } }); render(); - setMain(search, sections); + const tools = el("div", { class: "toolbar" }, + el("a", { class: "btn link-btn", href: "#/queues", text: "Queues" })); + setMain(tools, search, sections); } // ---- device: an assembly or a single item ----------------------------------------------------- @@ -1127,6 +1131,149 @@ async function showCamera(name) { if (document.visibilityState === "visible") start(); } +// ---- task queues: what runs and waits, pause/resume/stop ----------------------------------------------- + +const QUEUE_POLL = 1000; // ms + +function duration(seconds) { + if (seconds === null || seconds === undefined) return ""; + const s = Math.max(0, Math.round(seconds)); + return s < 60 ? `${s} s` : s < 3600 ? `${Math.floor(s / 60)} min ${s % 60} s` : `${Math.floor(s / 3600)} h ${Math.floor((s % 3600) / 60)} min`; +} + +async function showQueues() { + const view = newView("queues", ""); + setTitle("Queues", true); + const box = el("div", { class: "queues" }, el("div", { class: "loading", text: "loading queues…" })); + setMain(box, el("p", { class: "doc", text: "The session's task queues (%queue, scans with queue=…). " + + "Controls act on what is already queued; nothing is added here." })); + const writes = !!S.hello.writes; + const act = (queue, action, task, question) => { + const go = () => request("queue_action", { queue, action, task }).then(load, (e) => toast(e.message, true)); + question ? confirmDialog(question, "", go) : go(); + }; + const button = (text, onclick, cls = "") => + el("button", { class: `btn small ${cls}`, text, onclick, disabled: !writes, title: writes ? "" : "writes are off in this web session" }); + + const taskLine = (q, t, kind) => { + const took = t.started ? duration((t.finished || serverNow()) - t.started) : ""; + const head = el("div", { class: "task-head" }, + el("span", { class: `task-state ${t.state}`, text: t.state }), + el("span", { class: "task-id", text: `#${t.id}` }), + t.description ? el("span", { class: "task-desc", text: t.description }) : null, + took ? el("span", { class: "task-time", text: took }) : null); + const node = el("div", { class: `task ${kind}` }, head, el("div", { class: "task-source", text: t.source })); + if (t.error) node.append(el("div", { class: "task-error", text: t.error })); + if (t.scan) { + const pct = t.scan.total ? (100 * t.scan.done) / t.scan.total : 0; + node.append(el("div", { class: "progress" }, el("div", { class: "bar", style: `width:${pct.toFixed(1)}%` })), + el("div", { class: "task-scan", text: `step ${t.scan.done} / ${t.scan.total}` + + (t.scan.stopping ? " · stopping" : t.scan.paused ? " · scan paused" : "") })); + } + const tools = el("div", { class: "task-tools" }); + if (kind !== "pending") tools.append(el("a", { class: "btn small link-btn", href: `#/qout/${encodeURIComponent(q.name)}/${t.id}`, text: "Output" })); + if (kind === "running") { + if (t.scan) tools.append(t.scan.paused ? button("Resume scan", () => act(q.name, "resume_running")) + : button("Pause scan", () => act(q.name, "pause_running"))); + tools.append(button("Stop task", () => act(q.name, "stop_running", undefined, + `Stop task #${t.id}? Its scan stops at the next step; the queue pauses.`), "stop")); + } else if (kind === "pending") { + tools.append(button("Top", () => act(q.name, "move_top", t.id)), button("↑", () => act(q.name, "move_up", t.id)), + button("Remove", () => act(q.name, "remove", t.id, `Remove task #${t.id} from queue ${q.name}?`), "stop")); + } + if (tools.childElementCount) node.append(tools); + return node; + }; + + const render = (data) => { + if (!data.queues.length) { + box.replaceChildren(el("div", { class: "empty", text: "No queues in this session yet." })); + return; + } + box.replaceChildren(...data.queues.map((q) => { + const state = q.running ? (q.paused ? "running, then paused" : "running") : q.paused ? "paused" : q.pending.length ? "starting" : "idle"; + const card = el("section", { class: "queue-card" }, + el("div", { class: "queue-head" }, + el("h2", { text: q.name }), el("span", { class: `badge q-${q.paused ? "paused" : q.running ? "running" : "idle"}`, text: state }), + q.paused ? button("Resume queue", () => act(q.name, "resume", undefined, q.pending.length ? `Resume queue ${q.name}? ${q.pending.length} waiting task(s) will start.` : null)) + : button("Pause queue", () => act(q.name, "pause")))); + if (q.running) card.append(taskLine(q, q.running, "running")); + if (q.pending.length) { + card.append(el("div", { class: "section-title", text: `Waiting (${q.pending.length})` }), + ...q.pending.map((t) => taskLine(q, t, "pending")), + el("div", { class: "task-tools" }, + button("Drop all waiting", () => act(q.name, "clear", undefined, `Drop all ${q.pending.length} waiting tasks of ${q.name}?`), "stop"))); + } + if (q.running || q.pending.length) { + card.append(el("div", { class: "task-tools" }, button("Stop everything", () => act(q.name, "stop", undefined, + `Stop the running task of ${q.name} and drop everything waiting?`), "stop"))); + } + if (q.history.length) { + const list = el("div", {}, ...q.history.map((t) => taskLine(q, t, "done"))); + card.append(el("details", { class: "hidden-items" }, el("summary", { text: `Finished (${q.history.length})` }), list)); + } + return card; + })); + }; + let busy = false; + const open = new Set(); // keep "Finished" folds open across refreshes + const load = async () => { + if (busy || S.view !== view || document.visibilityState !== "visible") return; + busy = true; + try { + box.querySelectorAll("section.queue-card").forEach((c) => { + if (c.querySelector("details")?.open) open.add(c.querySelector("h2").textContent); + else open.delete(c.querySelector("h2")?.textContent); + }); + const data = await request("queues"); + if (S.view !== view) return; + render(data); + box.querySelectorAll("section.queue-card").forEach((c) => { + const d = c.querySelector("details"); + if (d && open.has(c.querySelector("h2").textContent)) d.open = true; + }); + } catch (e) { + if (S.view === view) box.replaceChildren(el("div", { class: "empty", text: `could not read the queues: ${e.message}` })); + } finally { + busy = false; + } + }; + await load(); + view.timers.push(setInterval(load, QUEUE_POLL)); +} + +async function showTaskOutput(path) { + const slash = path.lastIndexOf("/"); + const queue = path.slice(0, slash), id = parseInt(path.slice(slash + 1), 10); + const view = newView("qout", path); + setTitle(`${queue} #${id}`, true); + const head = el("div", { class: "task-head" }); + const pre = el("pre", { class: "task-output", text: "" }); + const follow = el("label", { class: "follow" }, el("input", { type: "checkbox", checked: true }), " follow"); + setMain(head, pre, follow); + let offset = 0, busy = false; + const load = async () => { + if (busy || S.view !== view || document.visibilityState !== "visible") return; + busy = true; + try { + const r = await request("queue_output", { queue, task: id, since: offset }); + if (S.view !== view) return; + if (r.offset < offset || r.clipped) pre.textContent = r.clipped ? "… (older output not shown)\n" : ""; + offset = r.offset; + if (r.text) pre.append(r.text); + head.replaceChildren(el("span", { class: `task-state ${r.task.state}`, text: r.task.state }), + el("span", { class: "task-source", text: r.task.source })); + if (follow.querySelector("input").checked && r.text) pre.scrollTop = pre.scrollHeight; + } catch (e) { + if (S.view === view) head.replaceChildren(el("span", { class: "task-error", text: e.message })); + } finally { + busy = false; + } + }; + await load(); + view.timers.push(setInterval(load, QUEUE_POLL)); +} + // ---- SVG panels ------------------------------------------------------------------------------------ const NAME_RE = /^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$/; diff --git a/eco/web/static/style.css b/eco/web/static/style.css index 14e355b..f40d57a 100644 --- a/eco/web/static/style.css +++ b/eco/web/static/style.css @@ -189,3 +189,29 @@ a.btn:not(.primary) { color: var(--link); } .cam-box { display: none; position: absolute; border: 2px dashed #fff; background: rgba(255,255,255,.12); pointer-events: none; } .cam-tools { margin-top: 8px; } .btn.on { background: var(--busy); color: #000; border-color: var(--busy); } + +/* task queues */ +.queue-card { background: var(--surface); border: 1px solid var(--line); border-radius: 12px; padding: 10px 12px; margin-bottom: 12px; } +.queue-head { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; } +.queue-head h2 { margin: 0; font-size: 18px; flex: 1; } +.badge.q-running { background: var(--busy); color: #000; } +.badge.q-paused { background: var(--err); color: #fff; } +.badge.q-idle { background: var(--line); color: var(--text); } +.task { border-top: 1px solid var(--line); padding: 8px 0; } +.task-head { display: flex; flex-wrap: wrap; align-items: baseline; gap: 4px 10px; } +.task-state { font-size: 12px; font-weight: 700; text-transform: uppercase; color: var(--muted); } +.task-state.running { color: var(--busy); } +.task-state.done { color: var(--ok); } +.task-state.error { color: var(--err); } +.task-id { color: var(--muted); font-size: 13px; } +.task-desc { font-weight: 600; } +.task-time { color: var(--muted); font-size: 13px; margin-left: auto; } +.task-source { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 13px; overflow-wrap: anywhere; } +.task-error { color: var(--err); font-size: 13px; overflow-wrap: anywhere; } +.task-scan { color: var(--muted); font-size: 13px; } +.task-tools { display: flex; flex-wrap: wrap; gap: 6px; margin-top: 6px; } +.progress { height: 8px; border-radius: 4px; background: var(--line); overflow: hidden; margin-top: 6px; } +.progress .bar { height: 100%; background: var(--busy); } +.task-output { background: var(--surface); border: 1px solid var(--line); border-radius: 12px; padding: 10px; max-height: 65vh; overflow: auto; + font-size: 12px; white-space: pre-wrap; overflow-wrap: anywhere; } +.follow { display: inline-flex; gap: 6px; align-items: center; color: var(--muted); font-size: 14px; } diff --git a/eco/web/static/sw.js b/eco/web/static/sw.js index 4b0a60e..c1850eb 100644 --- a/eco/web/static/sw.js +++ b/eco/web/static/sw.js @@ -3,7 +3,7 @@ // SVG panels or the websocket -- a stale beamline value shown as current // would be worse than none. Network first, so a new client version is // picked up as soon as the gateway is reachable. -const CACHE = "eco-shell-v4"; +const CACHE = "eco-shell-v5"; const SHELL = ["/", "/static/app.js", "/static/strip.js", "/static/style.css", "/icon.svg", "/manifest.webmanifest"]; self.addEventListener("install", (event) => { diff --git a/tests/test_web_client_session.py b/tests/test_web_client_session.py index 46a470c..bb0320c 100644 --- a/tests/test_web_client_session.py +++ b/tests/test_web_client_session.py @@ -326,3 +326,102 @@ def test_ptz_needs_writes_and_passes_the_access_gate(client, writer, monkeypatch writer.call("describe", name="cam_demo") reply = writer.call("ptz", name="cam_demo", action="home") assert not reply["ok"] and "not your camera" in reply["error"]["message"] + + +# -- task queues ------------------------------------------------------------------------- + + +def _queue_client(writes): + from eco.web.demo import demo_queues + + ns = DemoNamespace() + c = Client(writes=writes) + c.session.close() + queues = demo_queues(ns) + from eco.web.client_session import ClientSession + + c.session = ClientSession(SessionBackend(ns, queues=queues), c.messages.put, writes=writes, poll_interval=0.05) + return c, queues + + +def _wait(predicate, seconds=5.0): + deadline = time.monotonic() + seconds + while not predicate() and time.monotonic() < deadline: + time.sleep(0.05) + return predicate() + + +def test_queues_list_running_waiting_and_scan_progress(): + c, queues = _queue_client(writes=True) + try: + data = c.call("queues")["data"]["queues"] + default = next(q for q in data if q["name"] == "default") + assert default["paused"] and default["running"] is None and len(default["pending"]) == 3 + assert default["pending"][0]["description"] == "x -1..1" + assert c.call("queue_action", queue="default", action="resume")["ok"] + assert _wait(lambda: (queues["default"].current is not None and queues["default"].current.scan is not None + and queues["default"].current.scan._values_done)) + running = next(q for q in c.call("queues")["data"]["queues"] if q["name"] == "default")["running"] + assert running["state"] == "running" and running["scan"]["total"] == 40 and running["scan"]["done"] >= 1 + # pause the scan, then stop the task: the queue pauses, the rest waits + assert c.call("queue_action", queue="default", action="pause_running")["ok"] + assert _wait(lambda: queues["default"].current.scan.is_paused()) + assert c.call("queue_action", queue="default", action="stop_running")["ok"] + assert _wait(lambda: queues["default"].current is None) + q = next(q for q in c.call("queues")["data"]["queues"] if q["name"] == "default") + assert q["paused"] and len(q["pending"]) == 2 and q["history"][0]["state"] == "done" + out = c.call("queue_output", queue="default", task=q["history"][0]["id"])["data"] + assert "stopped on request" in out["text"] and out["offset"] == len(out["text"]) + again = c.call("queue_output", queue="default", task=q["history"][0]["id"], since=out["offset"])["data"] + assert again["text"] == "" + finally: + for queue in queues.values(): + queue.stop() + queue.close() + c.session.close() + + +def test_waiting_tasks_can_be_moved_and_removed(): + c, queues = _queue_client(writes=True) + try: + ids = [t.id for t in queues["default"].pending] + assert c.call("queue_action", queue="default", action="move_top", task=ids[2])["ok"] + assert [t.id for t in queues["default"].pending] == [ids[2], ids[0], ids[1]] + assert c.call("queue_action", queue="default", action="move_up", task=ids[1])["ok"] + assert [t.id for t in queues["default"].pending] == [ids[2], ids[1], ids[0]] + assert c.call("queue_action", queue="default", action="remove", task=ids[1])["ok"] + assert [t.id for t in queues["default"].pending] == [ids[2], ids[0]] + assert c.call("queue_action", queue="default", action="clear")["ok"] + assert queues["default"].pending == [] + reply = c.call("queue_action", queue="default", action="remove", task=ids[0]) + assert not reply["ok"] and "not waiting" in reply["error"]["message"] + assert not c.call("queue_action", queue="default", action="append")["ok"] + assert not c.call("queue_action", queue="nope", action="pause")["ok"] + assert "nope" not in queues + finally: + for queue in queues.values(): + queue.close() + c.session.close() + + +def test_queue_controls_need_writes(): + c, queues = _queue_client(writes=False) + try: + assert c.call("queues")["ok"] + reply = c.call("queue_action", queue="default", action="resume") + assert not reply["ok"] and reply["error"]["type"] == "WritesDisabled" + assert queues["default"].paused + finally: + for queue in queues.values(): + queue.close() + c.session.close() + + +def test_task_lines_name_devices_instead_of_their_status_tables(): + from eco.utilities.task_queue import Task + from eco.web.queues import task_source + + ns = DemoNamespace() + stage = ns.resolve_item("stage") + task = Task(lambda *a, **k: None, (stage.x, -1, [1, 2]), {"counters": [stage.diode], "step_info": None}) + assert task_source(task) == "test_task_lines_name_devices_instead_of_their_status_tables..(stage.x, -1, [1, 2], counters=[stage.diode], step_info=None)" diff --git a/web_frontend/PARITY.md b/web_frontend/PARITY.md index 8c16b51..3bbc64e 100644 --- a/web_frontend/PARITY.md +++ b/web_frontend/PARITY.md @@ -17,7 +17,9 @@ opened as a view; SVG panels; details (the grids' ⋯ menu); pinned live strip charts with running average; a picker over the initialized components; a tweak view (one axis); a **jog** view for hardware jogs (motor records: hold to move, jog speed ÷2/×2/set, dead-man stop in the gateway); a live view -of MJPEG PTZ network cameras (cam_west & co.) with tap-to-center/drag-to-zoom. Started +of MJPEG PTZ network cameras (cam_west & co.) with tap-to-center/drag-to-zoom; +the task queues with scan progress, output and the queue panel's controls +(not the launcher). Started with `eco web -s ` or `eco.start_web()`. ## Tier 1 — core, port first diff --git a/web_frontend/PROTOCOL.md b/web_frontend/PROTOCOL.md index d4c38d0..440c6c3 100644 --- a/web_frontend/PROTOCOL.md +++ b/web_frontend/PROTOCOL.md @@ -49,9 +49,9 @@ The client never sends expressions. | — | — | a camera's live stream is plain HTTP: `GET /api/mjpeg?name=…&quality=low|medium|full` returns `multipart/x-mixed-replace` JPEGs for an ``; `describe` of such an item has `camera: {stream: "mjpeg", ptz, rotate, qualities}` | | `jog_speed` | `{name, value}` | sets the jog speed of `name` (its `jog_speed` item); allowed whenever writing `name` is | | — | — | SVG panels are plain HTTP: `GET /api/svg?name=…` returns the SVG, header `X-Eco-Prefix` is what click markers are relative to; the client re-fetches every 3 s | -| `queues` *planned* | `{}` | queue list with tasks (mode B) | -| `queue_call` *planned* | `{queue, action: "pause"|"resume"|"abort_current"|…}` | allow-listed subset of `TaskQueue` methods | -| `queue_follow` *planned* | `{queue, from_task?}` | backlog + live `ev: "queue_output"` | +| `queues` | `{}` | `{queues: [{name, paused, running: task?, pending: [task], history: [task] (newest first, 10)}], now}`; task = `{id, source, description, state, started, finished, error?, scan?: {done, total, paused, stopping}}` — the client polls it every second | +| `queue_action` | `{queue, action, task?}` | (writes on, all items) `pause`, `resume`, `clear`, `stop`, `pause_running`, `resume_running`, `stop_running`, and for a waiting task `remove`, `move_up`, `move_top` (`eco.web.queues.ACTIONS`); nothing adds or repeats a task | +| `queue_output` | `{queue, task, since}` | `{text, offset, clipped, task}` — what the task printed after character `since` (at most the newest 20 000 characters) | | `tweak` *planned* | `{session_id, action, args}` | `TweakController` methods by allow-list | ## Events diff --git a/web_frontend/README.md b/web_frontend/README.md index 6377b37..ba11a33 100644 --- a/web_frontend/README.md +++ b/web_frontend/README.md @@ -46,6 +46,9 @@ phone size): (30 s .. 1 h) with optional running average, survive a reload - picker: find any readable item of the initialized components by name - tweak view: big step buttons, ×2/÷2, go to, back to start, stop, arrow keys +- task queues: running task with scan progress, waiting and finished tasks, + live task output; pause/resume queue and running scan, stop, move/remove a + waiting task, drop all (`eco/web/queues.py`; nothing adds or repeats a task) - live view of MJPEG network cameras (AxisPTZ: cam_west, ...), relayed by the gateway (`eco/web/camera.py`: one camera connection per quality, newest frame per client), with tap-to-center, drag-to-zoom, zoom steps, wide, home. Tested @@ -54,7 +57,7 @@ phone size): - token auth + HttpOnly cookie, binds 127.0.0.1, ping/reconnect after network drops, clean shutdown, service worker (app shell only), light/dark -Not yet: queues/scan progress, archiver history, memories, cam_server cameras, alarms, +Not yet: scan launcher (adding tasks), archiver history, memories, cam_server cameras, alarms, status-server (shared, read-only) mode. See [PARITY.md](PARITY.md) for the order. Points to keep in mind (found while wiring it in):