villekuosmanen commited on
Commit
b040ee7
·
verified ·
1 Parent(s): a191eda

deploy demo space from af5c03ea1d8d

Browse files
Files changed (2) hide show
  1. app.py +21 -5
  2. runner.py +82 -12
app.py CHANGED
@@ -276,6 +276,7 @@ _CSS = """
276
  .cell-dot--green { background: #22c55e; box-shadow: 0 0 7px 1px #22c55e; animation: cell-blink 1.4s ease-in-out infinite; }
277
  .cell-dot--yellow { background: #eab308; box-shadow: 0 0 7px 1px #eab308; animation: cell-blink 1.0s ease-in-out infinite; }
278
  .cell-dot--red { background: #ef4444; box-shadow: 0 0 7px 1px #ef4444; animation: cell-blink 0.8s ease-in-out infinite; }
 
279
  @keyframes cell-blink { 0%, 100% { opacity: 1; } 50% { opacity: 0.25; } }
280
  """
281
 
@@ -288,6 +289,10 @@ def _cell_light_html(snap: dict, task: config.Task) -> str:
288
  snap.get("cell_statuses") or {},
289
  aggregate=snap.get("cell_status", {}).get(task.slug, "offline"),
290
  )
 
 
 
 
291
  return (
292
  '<div class="cell-status">'
293
  f'<span class="cell-dot cell-dot--{color}"></span>'
@@ -420,9 +425,9 @@ with gr.Blocks(title="armnet eval — evaluate a policy on real robots") as demo
420
  "lerobot-eval runtime image ref) in the Space settings — see the README."
421
  )
422
 
423
- cell_light = gr.HTML(
424
- _cell_light_html({"cell_status": {}}, _DEFAULT_TASK)
425
- )
426
 
427
  with gr.Row():
428
  preset = gr.Dropdown(
@@ -518,8 +523,14 @@ with gr.Blocks(title="armnet eval — evaluate a policy on real robots") as demo
518
  inputs=[task_sel, preset, custom],
519
  outputs=[banner, status, run_btn, feed_sel, feed_caption, cell_light],
520
  )
 
 
 
521
  text_timer.tick(
522
- _refresh_text, inputs=[task_sel, feed_sel], outputs=text_outputs
 
 
 
523
  )
524
  feed_sel.change(
525
  _on_feed_change,
@@ -527,7 +538,12 @@ with gr.Blocks(title="armnet eval — evaluate a policy on real robots") as demo
527
  outputs=[status, rollouts, summary, events, feed_caption],
528
  )
529
 
530
- demo.load(_refresh_text, inputs=[task_sel, feed_sel], outputs=text_outputs)
 
 
 
 
 
531
  demo.load(
532
  _stream_video,
533
  outputs=video,
 
276
  .cell-dot--green { background: #22c55e; box-shadow: 0 0 7px 1px #22c55e; animation: cell-blink 1.4s ease-in-out infinite; }
277
  .cell-dot--yellow { background: #eab308; box-shadow: 0 0 7px 1px #eab308; animation: cell-blink 1.0s ease-in-out infinite; }
278
  .cell-dot--red { background: #ef4444; box-shadow: 0 0 7px 1px #ef4444; animation: cell-blink 0.8s ease-in-out infinite; }
279
+ .cell-dot--grey { background: #9ca3af; }
280
  @keyframes cell-blink { 0%, 100% { opacity: 1; } 50% { opacity: 0.25; } }
281
  """
282
 
 
289
  snap.get("cell_statuses") or {},
290
  aggregate=snap.get("cell_status", {}).get(task.slug, "offline"),
291
  )
292
+ return _light_html(color, caption)
293
+
294
+
295
+ def _light_html(color: str, caption: str) -> str:
296
  return (
297
  '<div class="cell-status">'
298
  f'<span class="cell-dot cell-dot--{color}"></span>'
 
425
  "lerobot-eval runtime image ref) in the Space settings — see the README."
426
  )
427
 
428
+ # Not "cells offline": a page that never gets its first refresh
429
+ # would then look exactly like a fleet that is down.
430
+ cell_light = gr.HTML(_light_html("grey", "checking cells…"))
431
 
432
  with gr.Row():
433
  preset = gr.Dropdown(
 
523
  inputs=[task_sel, preset, custom],
524
  outputs=[banner, status, run_btn, feed_sel, feed_caption, cell_light],
525
  )
526
+ # concurrency_limit=None on both: they share one Gradio concurrency group
527
+ # (keyed by the function), and its default limit of 1 would put every
528
+ # viewer's refresh behind a single worker. It only reads a locked snapshot.
529
  text_timer.tick(
530
+ _refresh_text,
531
+ inputs=[task_sel, feed_sel],
532
+ outputs=text_outputs,
533
+ concurrency_limit=None,
534
  )
535
  feed_sel.change(
536
  _on_feed_change,
 
538
  outputs=[status, rollouts, summary, events, feed_caption],
539
  )
540
 
541
+ demo.load(
542
+ _refresh_text,
543
+ inputs=[task_sel, feed_sel],
544
+ outputs=text_outputs,
545
+ concurrency_limit=None,
546
+ )
547
  demo.load(
548
  _stream_video,
549
  outputs=video,
runner.py CHANGED
@@ -24,7 +24,14 @@ from typing import Any, Optional
24
  import certifi
25
  import websocket
26
 
27
- from armnet_core import API_KEY_HEADER, Job, JobSpec, JobStatus, TerminalStatus
 
 
 
 
 
 
 
28
  from armnet_client._config import api_key, orchestrator_url
29
  from armnet_client.client import OrchestratorClient, OrchestratorError
30
  from armnet_client.execute import _ws_url
@@ -109,6 +116,33 @@ def fleet_availability(
109
  return "red", "cells offline"
110
 
111
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
112
  @dataclass
113
  class RunSlot:
114
  """Independent live state for one BusyBox cell's current/recent eval."""
@@ -544,28 +578,64 @@ class DemoState:
544
  previous: dict[str, str] = {}
545
  while True:
546
  for task in config.TASKS.values():
547
- status, reason, client, cells = self._poll_one(client, task)
 
 
 
 
 
 
 
 
548
  # Logged on change rather than every poll: at a few seconds per
549
  # poll this would otherwise be the loudest thing in the Space
550
  # log, but without it a red light gives an operator nothing to
551
  # act on.
552
- shown = self._apply_poll(task.slug, status, cells)
553
  if shown != previous.get(task.slug) and reason and shown == "offline":
554
  logger.warning("%s shown as offline: %s", task.label, reason)
555
  previous[task.slug] = shown
556
- try:
557
- self._refresh_queue(client)
558
- self._adopt_dispatched_eval(client)
559
- except Exception as exc: # noqa: BLE001 - keep last good queue view
560
- logger.warning("queue status poll failed: %r", exc)
561
  time.sleep(config.CELL_STATUS_POLL_S)
562
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
563
  def _apply_poll(
564
  self, task_slug: str, status: str, cells: Optional[list[Any]]
565
  ) -> str:
566
- """Record one poll. A failed poll keeps the last good fleet for a while."""
 
 
 
 
 
567
  with self._lock:
568
- if cells is None:
569
  fails = self._status_failures.get(task_slug, 0) + 1
570
  self._status_failures[task_slug] = fails
571
  if fails < _STATUS_FAILURES_BEFORE_OFFLINE:
@@ -588,8 +658,8 @@ class DemoState:
588
  orchestrator_url(), timeout=_STATUS_CALL_TIMEOUT_S
589
  )
590
  summary = _call_bounded(
591
- lambda: client.get_cell_status(
592
- embodiment=config.EMBODIMENT, task=task.slug
593
  ),
594
  _STATUS_CALL_TIMEOUT_S,
595
  )
 
24
  import certifi
25
  import websocket
26
 
27
+ from armnet_core import (
28
+ API_KEY_HEADER,
29
+ CellStatusSummary,
30
+ Job,
31
+ JobSpec,
32
+ JobStatus,
33
+ TerminalStatus,
34
+ )
35
  from armnet_client._config import api_key, orchestrator_url
36
  from armnet_client.client import OrchestratorClient, OrchestratorError
37
  from armnet_client.execute import _ws_url
 
116
  return "red", "cells offline"
117
 
118
 
119
+ def _read_cell_status(
120
+ client: Any, *, embodiment: str, task: str
121
+ ) -> CellStatusSummary:
122
+ """Fetch cell status without each cell's ``device_config``.
123
+
124
+ The Space never reads the device config, and it is the one strict part of
125
+ the response: the Space installs armnet-core from PyPI, which lags the
126
+ deployed orchestrator, and a config field added since then fails the whole
127
+ response. Every poll then fails and the lights stay offline until the
128
+ Space is rebuilt against a newer armnet-core.
129
+
130
+ Test doubles without an HTTP session keep using ``get_cell_status``.
131
+ """
132
+ http = getattr(client, "_client", None)
133
+ if http is None:
134
+ return client.get_cell_status(embodiment=embodiment, task=task)
135
+ response = http.get(
136
+ "/cells/status", params={"embodiment": embodiment, "task": task}
137
+ )
138
+ response.raise_for_status()
139
+ payload = response.json()
140
+ for cell in payload.get("cells") or []:
141
+ if isinstance(cell, dict):
142
+ cell.pop("device_config", None)
143
+ return CellStatusSummary.model_validate(payload)
144
+
145
+
146
  @dataclass
147
  class RunSlot:
148
  """Independent live state for one BusyBox cell's current/recent eval."""
 
578
  previous: dict[str, str] = {}
579
  while True:
580
  for task in config.TASKS.values():
581
+ # This is the only thread that updates the lights and the
582
+ # queue. If it dies they freeze for every viewer until the
583
+ # Space restarts.
584
+ try:
585
+ status, reason, client, cells = self._poll_one(client, task)
586
+ shown = self._apply_poll(task.slug, status, cells)
587
+ except Exception: # noqa: BLE001
588
+ logger.exception("%s status poll crashed", task.label)
589
+ continue
590
  # Logged on change rather than every poll: at a few seconds per
591
  # poll this would otherwise be the loudest thing in the Space
592
  # log, but without it a red light gives an operator nothing to
593
  # act on.
 
594
  if shown != previous.get(task.slug) and reason and shown == "offline":
595
  logger.warning("%s shown as offline: %s", task.label, reason)
596
  previous[task.slug] = shown
597
+ client = self._poll_queue(client)
 
 
 
 
598
  time.sleep(config.CELL_STATUS_POLL_S)
599
 
600
+ def _poll_queue(
601
+ self, client: Optional[OrchestratorClient]
602
+ ) -> Optional[OrchestratorClient]:
603
+ """Refresh the queue view and adopt started evals; the client to reuse.
604
+
605
+ Runs even when the status polls dropped their client. Otherwise the
606
+ queued count only grows (submissions add to it locally), a policy that
607
+ already started stays "already queued", and a full count disables Run.
608
+ """
609
+ try:
610
+ if client is None:
611
+ client = OrchestratorClient(
612
+ orchestrator_url(), timeout=_STATUS_CALL_TIMEOUT_S
613
+ )
614
+ _call_bounded(lambda: self._refresh_queue(client), _STATUS_CALL_TIMEOUT_S)
615
+ _call_bounded(
616
+ lambda: self._adopt_dispatched_eval(client), _STATUS_CALL_TIMEOUT_S
617
+ )
618
+ return client
619
+ except Exception as exc: # noqa: BLE001 - keep last good queue view
620
+ logger.warning("queue status poll failed: %r", exc)
621
+ if client is not None:
622
+ try:
623
+ client.close()
624
+ except Exception: # noqa: BLE001
625
+ pass
626
+ return None
627
+
628
  def _apply_poll(
629
  self, task_slug: str, status: str, cells: Optional[list[Any]]
630
  ) -> str:
631
+ """Record one poll. A failed or empty poll keeps the last good fleet a while.
632
+
633
+ An empty fleet is debounced too: an orchestrator instance that has just
634
+ started (a deploy, or Cloud Run scaling out) answers with no cells until
635
+ the first heartbeats reach it a few seconds later.
636
+ """
637
  with self._lock:
638
+ if not cells:
639
  fails = self._status_failures.get(task_slug, 0) + 1
640
  self._status_failures[task_slug] = fails
641
  if fails < _STATUS_FAILURES_BEFORE_OFFLINE:
 
658
  orchestrator_url(), timeout=_STATUS_CALL_TIMEOUT_S
659
  )
660
  summary = _call_bounded(
661
+ lambda: _read_cell_status(
662
+ client, embodiment=config.EMBODIMENT, task=task.slug
663
  ),
664
  _STATUS_CALL_TIMEOUT_S,
665
  )