url/homelable
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(zigbee): import as background scan run

Browse Source 
Mirrors IP scan flow: POST /zigbee/import-pending now creates a
ScanRun(kind=zigbee, status=running) and returns immediately.
Networkmap fetch + pending upsert run in the background, status
transitions to done/error when finished.

Frontend: import modal closes on submit, scan history shows the
run with a ZIG/IP kind chip and toasts on completion. Pending
modal auto-refreshes when run finishes.

scan_runs.kind column added (default 'ip', idempotent migration).
Existing zigbee tests refactored to exercise _persist_pending_import
directly (background tasks don't see the test session); route test
verifies the run is created with kind=zigbee.
This commit is contained in:
Pouzor
2026-05-10 00:45:21 +02:00
parent b17299f531
commit 0863c2db94
10 changed files with 143 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/zigbee.py
+51 -38
View File
@@ -1,16 +1,18 @@
"""FastAPI router for Zigbee2MQTT import."""
import logging
from datetime import datetime, timezone
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from sqlalchemy import delete as sa_delete
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.db.database import get_db
from app.db.models import Node, PendingDevice, PendingDeviceLink
from app.db.database import AsyncSessionLocal, get_db
from app.db.models import Node, PendingDevice, PendingDeviceLink, ScanRun
from app.schemas.scan import ScanRunResponse
from app.schemas.zigbee import (
ZigbeeCoordinatorOut,
ZigbeeEdgeOut,
@@ -66,48 +68,59 @@ async def import_zigbee_network(
return ZigbeeImportResponse(nodes=nodes, edges=edges, device_count=len(nodes))
@router.post("/import-pending", response_model=ZigbeeImportPendingResponse)
@router.post("/import-pending", response_model=ScanRunResponse)
async def import_zigbee_to_pending(
payload: ZigbeeImportRequest,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db),
_: str = Depends(get_current_user),
) -> ZigbeeImportPendingResponse:
"""Fetch the Z2M networkmap and store devices in the pending section.
) -> ScanRun:
"""Queue a Zigbee2MQTT pending import as a background scan run.
Coordinator is auto-approved (creates a canvas Node directly with
``ieee_address`` set). Routers and end devices are upserted into
``pending_devices`` keyed by IEEE address. The discovered parent→child
edges are persisted as ``pending_device_links`` rows so that approving a
pending device later can auto-create the corresponding Edge when the
other endpoint already exists as a canvas Node.
Re-importing replaces all zigbee-discovered links and updates pending
rows in place; pending devices not present in the new map are kept
untouched (the user may be mid-approval).
Returns the ScanRun row immediately so the UI can close the import
modal and surface progress under Scan History (kind=zigbee). The
actual MQTT fetch + pending upsert happens in the background.
"""
try:
nodes_raw, edges_raw = await fetch_networkmap(
mqtt_host=payload.mqtt_host,
mqtt_port=payload.mqtt_port,
base_topic=payload.base_topic,
username=payload.mqtt_username,
password=payload.mqtt_password,
tls=payload.mqtt_tls,
tls_insecure=payload.mqtt_tls_insecure,
)
except ImportError as exc:
raise HTTPException(status_code=500, detail=str(exc)) from exc
except ConnectionError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except TimeoutError as exc:
raise HTTPException(status_code=504, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except Exception as exc:
logger.exception("Unexpected error during Zigbee pending import")
raise HTTPException(status_code=500, detail="Unexpected error during Zigbee import") from exc
run = ScanRun(
status="running",
kind="zigbee",
ranges=[f"{payload.mqtt_host}:{payload.mqtt_port}"],
)
db.add(run)
await db.commit()
await db.refresh(run)
background_tasks.add_task(_background_zigbee_import, run.id, payload)
return run
return await _persist_pending_import(db, nodes_raw, edges_raw)
async def _background_zigbee_import(run_id: str, payload: ZigbeeImportRequest) -> None:
async with AsyncSessionLocal() as db:
try:
nodes_raw, edges_raw = await fetch_networkmap(
mqtt_host=payload.mqtt_host,
mqtt_port=payload.mqtt_port,
base_topic=payload.base_topic,
username=payload.mqtt_username,
password=payload.mqtt_password,
tls=payload.mqtt_tls,
tls_insecure=payload.mqtt_tls_insecure,
)
result = await _persist_pending_import(db, nodes_raw, edges_raw)
run = await db.get(ScanRun, run_id)
if run:
run.status = "done"
run.devices_found = result.device_count
run.finished_at = datetime.now(timezone.utc)
await db.commit()
except Exception as exc:
logger.exception("Zigbee import %s failed", run_id)
await db.rollback()
run = await db.get(ScanRun, run_id)
if run:
run.status = "error"
run.error = str(exc)[:500]
run.finished_at = datetime.now(timezone.utc)
await db.commit()
async def _persist_pending_import(
backend/app/db/database.py
+2
View File
@@ -97,6 +97,8 @@ async def init_db() -> None:
await conn.exec_driver_sql("ALTER TABLE nodes ADD COLUMN bottom_handles INTEGER NOT NULL DEFAULT 1")
with suppress(OperationalError):
await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN discovery_source TEXT")
with suppress(OperationalError):
await conn.exec_driver_sql("ALTER TABLE scan_runs ADD COLUMN kind TEXT NOT NULL DEFAULT 'ip'")
# --- Zigbee schema migrations (logged variant per CLAUDE.md feedback) ---
zigbee_migrations: list[tuple[str, str]] = [
("nodes.ieee_address", "ALTER TABLE nodes ADD COLUMN ieee_address TEXT"),
backend/app/db/models.py
+1
View File
@@ -128,6 +128,7 @@ class ScanRun(Base):
id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid)
status: Mapped[str] = mapped_column(String, default="running")
kind: Mapped[str] = mapped_column(String, default="ip", server_default="ip")
ranges: Mapped[list[str]] = mapped_column(JSON, default=list)
devices_found: Mapped[int] = mapped_column(Integer, default=0)
started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
backend/app/schemas/scan.py
+1
View File
@@ -28,6 +28,7 @@ class PendingDeviceResponse(BaseModel):
class ScanRunResponse(BaseModel):
id: str
status: str
kind: str = "ip"
ranges: list[str]
devices_found: int
started_at: datetime
backend/tests/test_zigbee_router.py
+49 -77
View File
@@ -314,106 +314,78 @@ _PENDING_EDGES = [
@pytest.mark.asyncio
async def test_import_pending_creates_coordinator_and_pending(
async def test_import_pending_endpoint_creates_zigbee_scan_run(
client: AsyncClient, headers: dict
) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = (_PENDING_NODES, _PENDING_EDGES)
"""Endpoint returns a ScanRun (kind=zigbee, status=running) immediately;
the actual networkmap fetch + pending persist runs in the background."""
from unittest.mock import AsyncMock
with patch(
"app.api.routes.zigbee._background_zigbee_import",
new_callable=AsyncMock,
):
res = await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
assert res.status_code == 200
data = res.json()
assert data["device_count"] == 3
assert data["pending_created"] == 2 # router + enddevice
assert data["pending_updated"] == 0
assert data["coordinator"] is not None
assert data["coordinator"]["ieee_address"] == "0xCOORD"
assert data["coordinator_already_existed"] is False
assert data["links_recorded"] == 2
pending = await client.get("/api/v1/scan/pending", headers=headers)
assert pending.status_code == 200
rows = pending.json()
ieees = {r["ieee_address"] for r in rows}
assert ieees == {"0xR1", "0xE1"}
router = next(r for r in rows if r["ieee_address"] == "0xR1")
assert router["model"] == "CC2530"
assert router["lqi"] == 220
assert router["device_subtype"] == "Router"
assert router["discovery_source"] == "zigbee"
run = res.json()
assert run["kind"] == "zigbee"
assert run["status"] == "running"
assert run["ranges"] == ["localhost:1883"]
@pytest.mark.asyncio
async def test_import_pending_idempotent_updates_existing(
client: AsyncClient, headers: dict
async def test_persist_pending_import_creates_coordinator_and_pending(
db_session,
) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = (_PENDING_NODES, _PENDING_EDGES)
await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
from app.api.routes.zigbee import _persist_pending_import
bumped = [dict(n) for n in _PENDING_NODES]
bumped[1]["lqi"] = 99
res = await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
# second call: returns the bumped data
mock_fetch.return_value = (bumped, _PENDING_EDGES)
res = await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
assert res.status_code == 200
data = res.json()
assert data["pending_created"] == 0
assert data["pending_updated"] == 2
assert data["coordinator_already_existed"] is True
assert data["links_recorded"] == 2
pending = await client.get("/api/v1/scan/pending", headers=headers)
router = next(r for r in pending.json() if r["ieee_address"] == "0xR1")
assert router["lqi"] == 99
result = await _persist_pending_import(db_session, _PENDING_NODES, _PENDING_EDGES)
assert result.device_count == 3
assert result.pending_created == 2
assert result.pending_updated == 0
assert result.coordinator is not None
assert result.coordinator.ieee_address == "0xCOORD"
assert result.coordinator_already_existed is False
assert result.links_recorded == 2
@pytest.mark.asyncio
async def test_import_pending_replaces_links(
client: AsyncClient, headers: dict, db_session
async def test_persist_pending_import_idempotent_updates_existing(
db_session,
) -> None:
"""Re-importing wipes old zigbee links and inserts only the fresh set."""
from app.api.routes.zigbee import _persist_pending_import
await _persist_pending_import(db_session, _PENDING_NODES, _PENDING_EDGES)
bumped = [dict(n) for n in _PENDING_NODES]
bumped[1]["lqi"] = 99
result = await _persist_pending_import(db_session, bumped, _PENDING_EDGES)
assert result.pending_created == 0
assert result.pending_updated == 2
assert result.coordinator_already_existed is True
assert result.links_recorded == 2
@pytest.mark.asyncio
async def test_persist_pending_import_replaces_links(db_session) -> None:
from sqlalchemy import select
from app.api.routes.zigbee import _persist_pending_import
from app.db.models import PendingDeviceLink
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = (_PENDING_NODES, _PENDING_EDGES)
await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
await _persist_pending_import(db_session, _PENDING_NODES, _PENDING_EDGES)
new_edges = [{"source": "0xCOORD", "target": "0xR1"}]
mock_fetch.return_value = (_PENDING_NODES[:2], new_edges)
await client.post(
"/api/v1/zigbee/import-pending",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
new_edges = [{"source": "0xCOORD", "target": "0xR1"}]
await _persist_pending_import(db_session, _PENDING_NODES[:2], new_edges)
result = await db_session.execute(select(PendingDeviceLink))
links = result.scalars().all()
assert len(links) == 1
assert (links[0].source_ieee, links[0].target_ieee) == ("0xCOORD", "0xR1")
rows = (await db_session.execute(select(PendingDeviceLink))).scalars().all()
assert len(rows) == 1
assert (rows[0].source_ieee, rows[0].target_ieee) == ("0xCOORD", "0xR1")
@pytest.mark.asyncio
frontend/src/App.tsx
+3 -19
View File
@@ -541,25 +541,9 @@ export default function App() {
open={zigbeeImportOpen}
onClose={() => setZigbeeImportOpen(false)}
onAddToCanvas={handleZigbeeAddToCanvas}
onPendingImported={(coordinator) => {
useCanvasStore.getState().notifyScanDeviceFound()
if (coordinator) {
const exists = useCanvasStore.getState().nodes.some((n) => n.id === coordinator.id)
if (!exists) {
addNode({
id: coordinator.id,
type: 'zigbee_coordinator',
position: { x: 600, y: 100 },
data: {
label: coordinator.label,
type: 'zigbee_coordinator' as NodeData['type'],
status: 'unknown' as const,
services: [],
},
})
markUnsaved()
}
}
onPendingImported={() => {
setSidebarForceView(undefined)
setTimeout(() => setSidebarForceView('history'), 0)
}}
/>
)}
frontend/src/api/client.ts
+8 -6
View File
@@ -125,11 +125,13 @@ export const zigbeeApi = {
mqtt_tls_insecure?: boolean
}) =>
api.post<{
pending_created: number
pending_updated: number
coordinator: { id: string; label: string; ieee_address: string } | null
coordinator_already_existed: boolean
links_recorded: number
device_count: number
id: string
status: string
kind: string
ranges: string[]
devices_found: number
started_at: string
finished_at: string | null
error: string | null
}>('/zigbee/import-pending', data),
}
frontend/src/components/panels/Sidebar.tsx
+17 -1
View File
@@ -20,6 +20,7 @@ const PENDING_TRIGGERS: { kind: 'pending' | 'hidden'; icon: typeof ScanLine; lab
interface ScanRun {
id: string
status: string
kind?: string
ranges: string[]
devices_found: number
started_at: string
@@ -197,12 +198,19 @@ function ScanHistoryPanel() {
const res = await scanApi.runs()
const next: ScanRun[] = res.data
// Toast when a run transitions from running → error
// Surface transitions and refresh dependent UI
for (const run of next) {
const prev = prevRunsRef.current.find((r) => r.id === run.id)
if (prev?.status === 'running' && run.status === 'error') {
toast.error(`Scan failed: ${run.error ?? 'unknown error'}`)
}
if (prev?.status === 'running' && run.status === 'done') {
if (run.kind === 'zigbee') {
toast.success(`Zigbee import done — ${run.devices_found} device${run.devices_found !== 1 ? 's' : ''}`)
}
// Notify pending modal/canvas to refresh
useCanvasStore.getState().notifyScanDeviceFound()
}
}
prevRunsRef.current = next
setRuns(next)
@@ -263,6 +271,14 @@ function ScanHistoryPanel() {
<span className="w-1.5 h-1.5 rounded-full shrink-0" style={{ backgroundColor: statusColor(r.status) }} />
<span className="font-mono text-foreground capitalize">{r.status}</span>
{r.status === 'running' && <Loader2 size={10} className="animate-spin text-[#e3b341]" />}
<span
className="text-[9px] font-mono px-1 py-0.5 rounded uppercase tracking-wider"
style={r.kind === 'zigbee'
? { background: '#00d4ff22', color: '#00d4ff' }
: { background: '#a855f722', color: '#a855f7' }}
>
{r.kind === 'zigbee' ? 'ZIG' : 'IP'}
</span>
<span className="ml-auto text-muted-foreground font-mono">{r.devices_found} found</span>
{r.status === 'running' && (
<Tooltip>
frontend/src/components/zigbee/ZigbeeImportModal.tsx
+3 -13
View File
@@ -138,19 +138,9 @@ export function ZigbeeImportModal({ open, onClose, onAddToCanvas, onPendingImpor
setLoading(true)
try {
if (importMode === 'pending') {
const res = await zigbeeApi.importToPending(buildPayload())
const { pending_created, pending_updated, coordinator, coordinator_already_existed, device_count } = res.data
if (device_count === 0) {
toast.info('No Zigbee devices found in the network map')
} else {
const coordMsg = coordinator_already_existed
? 'coordinator already on canvas'
: 'coordinator added to canvas'
toast.success(
`Imported ${pending_created} new, updated ${pending_updated} (${coordMsg})`,
)
}
onPendingImported?.(coordinator)
await zigbeeApi.importToPending(buildPayload())
toast.success('Zigbee import started — track progress in Scan History')
onPendingImported?.(null)
handleClose()
} else {
const res = await zigbeeApi.importNetwork(buildPayload())
frontend/src/components/zigbee/__tests__/ZigbeeImportModal.test.tsx
+8 -6
View File
@@ -181,12 +181,14 @@ describe('ZigbeeImportModal', () => {
it('imports to pending by default and notifies parent', async () => {
vi.mocked(zigbeeApi.importToPending).mockResolvedValue({
data: {
pending_created: 2,
pending_updated: 0,
coordinator: { id: 'coord-uuid', label: 'Coordinator', ieee_address: '0x0000' },
coordinator_already_existed: false,
links_recorded: 1,
device_count: 3,
id: 'run-1',
status: 'running',
kind: 'zigbee',
ranges: ['192.168.1.100:1883'],
devices_found: 0,
started_at: '2026-01-01T00:00:00Z',
finished_at: null,
error: null,
},
} as never)
const onPendingImported = vi.fn()