feat: show last seen from periodic fetch

chi_edge/cli.py

@@ -14,7 +14,7 @@
 import contextlib
 import json
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from uuid import UUID
 from typing import Any
@@ -239,9 +239,7 @@ def list_all(long_: "bool" = False):
                 device["uuid"],
                 localize(device["created_at"]),
                 registration_state,
-                localize(balena_worker["state_details"].get("last_seen", "--"))
-                if balena_worker
-                else "--",
+                _last_seen_cell(balena_worker),
             ]
             if long_:
                 projects = device["properties"].get("authorized_projects") or []
@@ -532,8 +530,8 @@ def print_device(hardware):
     workers = hardware["workers"]
     cols = [""] + [w["worker_type"] for w in workers]
     worker_table = make_table(*cols, header_style="blue")
-    for key in ["state", "state_details"]:
-        worker_table.add_row(*([key] + [format_value(w[key]) for w in workers]))
+    for key in ["state", "state_details", "observed_state"]:
+        worker_table.add_row(*([key] + [format_value(w.get(key)) for w in workers]))
     outer.add_row(worker_table)
     console.print(Panel(outer, title=title, title_align="left"))
 
@@ -569,8 +567,12 @@ def resolve_device(doni_client, device_ref: "str"):
 
 
 def parse_date(utc_datestr):
-    parsed_date = datetime.strptime(utc_datestr, "%Y-%m-%dT%H:%M:%S+00:00")
-    return parsed_date
+    if utc_datestr.endswith("Z"):
+        utc_datestr = utc_datestr[:-1] + "+00:00"
+    parsed = datetime.fromisoformat(utc_datestr)
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
 
 
 def localize(utc_datestr):
@@ -582,6 +584,33 @@ def localize(utc_datestr):
         return utc_datestr
 
 
+def humanize_delta(seconds: int) -> str:
+    if seconds < 60:
+        return f"{seconds}s"
+    if seconds < 3600:
+        return f"{seconds // 60}m"
+    if seconds < 86400:
+        h, m = divmod(seconds, 3600)
+        return f"{h}h {m // 60}m"
+    return f"{seconds // 86400}d"
+
+
+def _last_seen_cell(balena_worker):
+    if not balena_worker:
+        return "--"
+    observed = balena_worker.get("observed_state") or {}
+    if not observed:
+        return "--"
+    if observed.get("is_online"):
+        return "[green]online[/green]"
+    last_event = observed.get("last_connectivity_event")
+    if not last_event:
+        return "--"
+    delta = datetime.now(timezone.utc) - parse_date(last_event)
+    duration = humanize_delta(int(delta.total_seconds()))
+    return f"[red]offline for {duration}[/red]"
+
+
 def format_value(value):
     if isinstance(value, dict) or isinstance(value, list):
         return yaml.dump(value).strip()

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "python-chi-edge"
-version = "0.3.2"
+version = "0.3.3"
 description = "Manage edge devices for use with the CHI@Edge IoT/Edge testbed."
 authors = [{ name = "Chameleon Project", email = "contact@chameleoncloud.org" }]
 readme = "README.md"

tests/test_cli.py

@@ -1,3 +1,4 @@
+import copy
 from unittest.mock import patch, MagicMock
 
 from click.testing import CliRunner
@@ -26,6 +27,10 @@
                 "fleet_id": 1918419,
                 "last_seen": "2025-11-07T17:29:43+00:00",
             },
+            "observed_state": {
+                "is_online": True,
+                "last_connectivity_event": "2025-11-07T17:29:43.123Z",
+            },
         },
         {
             "worker_type": "blazar.device",
@@ -162,16 +167,54 @@ def test_device_list_long_shows_extra_columns():
         assert "allowed" in result.output
 
 
+def test_device_list_shows_online_status():
+    mock_adapter = MagicMock()
+    mock_adapter.get.return_value.json.return_value = {"hardware": [FAKE_DEVICE]}
+
+    runner = CliRunner()
+    with (
+        patch("chi_edge.cli.doni_client", return_value=mock_adapter),
+        patch("chi_edge.cli.console", Console(width=300)),
+    ):
+        result = runner.invoke(cli, ["device", "list"])
+        assert result.exit_code == 0, result.output
+        assert "online" in result.output
+
+
+def test_device_list_shows_offline_status():
+    device = copy.deepcopy(FAKE_DEVICE)
+    for w in device["workers"]:
+        if w["worker_type"] == "balena":
+            w["observed_state"]["is_online"] = False
+
+    mock_adapter = MagicMock()
+    mock_adapter.get.return_value.json.return_value = {"hardware": [device]}
+
+    runner = CliRunner()
+    with (
+        patch("chi_edge.cli.doni_client", return_value=mock_adapter),
+        patch("chi_edge.cli.console", Console(width=300)),
+    ):
+        result = runner.invoke(cli, ["device", "list"])
+        assert result.exit_code == 0, result.output
+        assert "offline for" in result.output
+
+
 def test_device_show():
     mock_adapter = MagicMock()
     mock_adapter.get.return_value.json.return_value = FAKE_DEVICE
 
     runner = CliRunner()
-    with patch("chi_edge.cli.doni_client", return_value=mock_adapter):
+    with (
+        patch("chi_edge.cli.doni_client", return_value=mock_adapter),
+        patch("chi_edge.cli.console", Console(width=300)),
+    ):
         result = runner.invoke(cli, ["device", "show", FAKE_DEVICE["uuid"]])
         assert result.exit_code == 0, result.output
         assert "iot-rpi4-01" in result.output
         assert "raspberrypi4-64" in result.output
+        assert "is_online" in result.output
+        assert "last_connectivity_event" in result.output
 
 
 def test_device_set_scalar():