feat: show inventory timestamps on Device Inventory tiles

Surface the same lifecycle timestamps on the Device Inventory cards as on the
detail panel. Tiles for devices placed on a canvas show their linked node's
created / last scan / last modified / last seen (correlated by ip or
ieee_address, aggregated across matches: created = oldest, others = newest).
Devices not yet on any canvas fall back to their discovered_at.

Rendered as compact relative times ("2d ago") with the full date on hover, in
a tight two-column footer so the tile keeps its original footprint.

Backend: PendingDeviceResponse gains node_created_at / node_last_scan /
node_last_modified / node_last_seen; the canvas correlation now also pulls node
timestamps in the same single query. Frontend: shared timeFormat util
(absolute + relative), reused by the detail panel.

ha-relevant: maybe
This commit is contained in:
Pouzor
2026-06-27 13:24:29 +02:00
parent 19b7d38ec0
commit 612280e924
9 changed files with 279 additions and 35 deletions
+57 -27
View File
@@ -1,6 +1,7 @@
import ipaddress
import logging
import uuid
from datetime import datetime
from typing import Any
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
@@ -193,51 +194,80 @@ async def stop_scan(
return {"stopping": True}
async def _canvas_counts(
db: AsyncSession, devices: list[PendingDevice]
) -> dict[str, int]:
"""Map each device id → number of distinct canvases (designs) it appears on.
def _agg(values: list[datetime], *, newest: bool) -> datetime | None:
"""Pick the newest (max) or oldest (min) of a list of timestamps, or None."""
present = [v for v in values if v is not None]
if not present:
return None
return max(present) if newest else min(present)
Correlates a scanned device to existing nodes by ``ieee_address`` (exact) or
``ip`` (exact). Runs a single node query and groups in Python — node counts are
small for a homelab, so this avoids an N+1 per device.
async def _canvas_correlation(
db: AsyncSession, devices: list[PendingDevice]
) -> dict[str, dict[str, Any]]:
"""Correlate each device to existing canvas nodes by ``ieee_address`` or ``ip``.
Returns, per device id: the number of distinct canvases (designs) it appears
on, plus aggregated timestamps from every matching node — created_at (oldest),
last_scan / updated_at / last_seen (newest). One node query, grouped in Python
(node counts are small for a homelab), so no N+1 per device.
"""
if not devices:
return {}
rows = (
await db.execute(
select(Node.ip, Node.ieee_address, Node.design_id).where(
Node.design_id.isnot(None)
)
select(
Node.ip,
Node.ieee_address,
Node.design_id,
Node.created_at,
Node.last_scan,
Node.updated_at,
Node.last_seen,
).where(Node.design_id.isnot(None))
)
).all()
# ip → set(design_id), ieee → set(design_id)
by_ip: dict[str, set[str]] = {}
by_ieee: dict[str, set[str]] = {}
for ip, ieee, design_id in rows:
if ip:
by_ip.setdefault(ip, set()).add(design_id)
if ieee:
by_ieee.setdefault(ieee, set()).add(design_id)
# Index matching nodes by ip and by ieee so a device can look up both.
by_ip: dict[str, list[Any]] = {}
by_ieee: dict[str, list[Any]] = {}
for row in rows:
if row.ip:
by_ip.setdefault(row.ip, []).append(row)
if row.ieee_address:
by_ieee.setdefault(row.ieee_address, []).append(row)
counts: dict[str, int] = {}
info: dict[str, dict[str, Any]] = {}
for d in devices:
designs: set[str] = set()
matched = []
if d.ieee_address:
designs |= by_ieee.get(d.ieee_address, set())
matched += by_ieee.get(d.ieee_address, [])
if d.ip:
designs |= by_ip.get(d.ip, set())
counts[d.id] = len(designs)
return counts
matched += by_ip.get(d.ip, [])
# De-duplicate nodes matched by both ip and ieee.
matched = list({id(m): m for m in matched}.values())
designs = {m.design_id for m in matched}
info[d.id] = {
"canvas_count": len(designs),
"node_created_at": _agg([m.created_at for m in matched], newest=False),
"node_last_scan": _agg([m.last_scan for m in matched], newest=True),
"node_last_modified": _agg([m.updated_at for m in matched], newest=True),
"node_last_seen": _agg([m.last_seen for m in matched], newest=True),
}
return info
async def _with_canvas_counts(
db: AsyncSession, devices: list[PendingDevice]
) -> list[PendingDevice]:
"""Attach a transient ``canvas_count`` to each device for the response schema."""
counts = await _canvas_counts(db, devices)
"""Attach transient canvas count + linked-node timestamps for the response."""
info = await _canvas_correlation(db, devices)
for d in devices:
d.canvas_count = counts.get(d.id, 0)
meta = info.get(d.id, {})
d.canvas_count = meta.get("canvas_count", 0)
d.node_created_at = meta.get("node_created_at")
d.node_last_scan = meta.get("node_last_scan")
d.node_last_modified = meta.get("node_last_modified")
d.node_last_seen = meta.get("node_last_seen")
return devices
+7
View File
@@ -24,6 +24,13 @@ class PendingDeviceResponse(BaseModel):
# Number of distinct canvases (designs) this device already appears on,
# correlated by ip / ieee_address against existing nodes. Computed per-request.
canvas_count: int = 0
# Timestamps from the linked canvas node(s), correlated by ip / ieee_address.
# Null when the device is not on any canvas yet. Aggregated across matches:
# created_at = oldest; last_scan / last_modified / last_seen = newest.
node_created_at: datetime | None = None
node_last_scan: datetime | None = None
node_last_modified: datetime | None = None
node_last_seen: datetime | None = None
model_config = {"from_attributes": True}
+51
View File
@@ -1,5 +1,6 @@
"""Tests for scan routes: trigger, pending devices, approve/hide/ignore, stop."""
import uuid
from datetime import datetime, timezone
from unittest.mock import AsyncMock, patch
import pytest
@@ -229,6 +230,56 @@ async def test_canvas_count_ignores_nodes_without_design(client, headers, db_ses
assert res.json()[0]["canvas_count"] == 0
# --- Linked-node timestamps on the inventory response ---
@pytest.mark.asyncio
async def test_pending_device_without_node_has_null_node_timestamps(client, headers, pending_device):
# No matching canvas node → node_* timestamps are all null; the device still
# carries its own discovered_at for the "Discovered" fallback on the tile.
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
assert data["discovered_at"] is not None
assert data["node_created_at"] is None
assert data["node_last_scan"] is None
assert data["node_last_modified"] is None
assert data["node_last_seen"] is None
@pytest.mark.asyncio
async def test_pending_device_exposes_linked_node_timestamps(client, headers, db_session, pending_device):
d1 = await _add_design(db_session, "Home")
node = _node(d1, ip="192.168.1.100")
node.last_scan = datetime(2026, 6, 1, 8, 30, tzinfo=timezone.utc)
node.last_seen = datetime(2026, 6, 25, 9, 15, tzinfo=timezone.utc)
db_session.add(node)
await db_session.commit()
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
assert data["node_created_at"] is not None # defaulted on insert
assert data["node_last_modified"] is not None # updated_at defaulted on insert
assert data["node_last_scan"].startswith("2026-06-01")
assert data["node_last_seen"].startswith("2026-06-25")
@pytest.mark.asyncio
async def test_node_timestamps_aggregate_across_matches(client, headers, db_session, pending_device):
# Two canvas nodes share the device IP: created_at takes the OLDEST,
# last_scan takes the NEWEST.
d1 = await _add_design(db_session, "Home")
d2 = await _add_design(db_session, "Lab")
older = _node(d1, ip="192.168.1.100")
older.created_at = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc)
older.last_scan = datetime(2026, 3, 1, 0, 0, tzinfo=timezone.utc)
newer = _node(d2, ip="192.168.1.100")
newer.created_at = datetime(2026, 5, 1, 0, 0, tzinfo=timezone.utc)
newer.last_scan = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
db_session.add_all([older, newer])
await db_session.commit()
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
assert data["node_created_at"].startswith("2026-01-01") # oldest
assert data["node_last_scan"].startswith("2026-06-01") # newest
# --- Approve device ---
@pytest.mark.asyncio