feat: Phase 6, Device provisioning and deployment of updates on git-pull

backend/manufacturing/service.py

@@ -6,7 +6,7 @@ from shared.firebase import get_db
 from shared.exceptions import NotFoundError
 from utils.serial_number import generate_serial
 from utils.nvs_generator import generate as generate_nvs_binary
-from manufacturing.models import BatchCreate, BatchResponse, DeviceInventoryItem, DeviceStatusUpdate
+from manufacturing.models import BatchCreate, BatchResponse, DeviceInventoryItem, DeviceStatusUpdate, DeviceAssign, ManufacturingStats, RecentActivityItem
 
 COLLECTION = "devices"
 _BATCH_ID_CHARS = string.ascii_uppercase + string.digits
@@ -151,3 +151,77 @@ def get_nvs_binary(sn: str) -> bytes:
         hw_type=item.hw_type,
         hw_version=item.hw_version,
     )
+
+
+def assign_device(sn: str, data: DeviceAssign) -> DeviceInventoryItem:
+    from utils.email import send_device_assignment_invite
+
+    db = get_db()
+    docs = list(db.collection(COLLECTION).where("serial_number", "==", sn).limit(1).stream())
+    if not docs:
+        raise NotFoundError("Device")
+
+    doc_ref = docs[0].reference
+    doc_ref.update({
+        "owner": data.customer_email,
+        "assigned_to": data.customer_email,
+        "mfg_status": "sold",
+    })
+
+    send_device_assignment_invite(
+        customer_email=data.customer_email,
+        serial_number=sn,
+        customer_name=data.customer_name,
+    )
+
+    return _doc_to_inventory_item(doc_ref.get())
+
+
+def get_stats() -> ManufacturingStats:
+    db = get_db()
+    docs = list(db.collection(COLLECTION).stream())
+
+    all_statuses = ["manufactured", "flashed", "provisioned", "sold", "claimed", "decommissioned"]
+    counts = {s: 0 for s in all_statuses}
+
+    activity_candidates = []
+    for doc in docs:
+        data = doc.to_dict() or {}
+        status = data.get("mfg_status", "manufactured")
+        if status in counts:
+            counts[status] += 1
+
+        if status in ("provisioned", "sold", "claimed"):
+            # Use created_at as a proxy timestamp; Firestore DatetimeWithNanoseconds or plain datetime
+            ts = data.get("created_at")
+            if isinstance(ts, datetime):
+                ts_str = ts.strftime("%Y-%m-%dT%H:%M:%SZ")
+            else:
+                ts_str = str(ts) if ts else None
+
+            activity_candidates.append(RecentActivityItem(
+                serial_number=data.get("serial_number", ""),
+                hw_type=data.get("hw_type", ""),
+                mfg_status=status,
+                owner=data.get("owner"),
+                updated_at=ts_str,
+            ))
+
+    # Sort by updated_at descending, take latest 10
+    activity_candidates.sort(
+        key=lambda x: x.updated_at or "",
+        reverse=True,
+    )
+    recent = activity_candidates[:10]
+
+    return ManufacturingStats(counts=counts, recent_activity=recent)
+
+
+def get_firmware_url(sn: str) -> str:
+    """Return the FastAPI download URL for the latest stable firmware for this device's hw_type."""
+    from firmware.service import get_latest
+    item = get_device_by_sn(sn)
+    hw_type = item.hw_type.lower()
+    latest = get_latest(hw_type, "stable")
+    # download_url is a relative path like /api/firmware/vs/stable/1.4.2/firmware.bin
+    return latest.download_url