77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter()
|
|
|
|
# In-memory store (demo — resets on restart)
|
|
_windows: list[dict] = []
|
|
|
|
|
|
class WindowCreate(BaseModel):
|
|
site_id: str
|
|
title: str
|
|
target: str # "all", a room_id like "hall-a", or a rack_id like "rack-A01"
|
|
target_label: str # human-readable label
|
|
start_dt: str # ISO 8601
|
|
end_dt: str # ISO 8601
|
|
suppress_alarms: bool = True
|
|
notes: str = ""
|
|
|
|
|
|
def _window_status(w: dict) -> str:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
if w["end_dt"] < now:
|
|
return "expired"
|
|
if w["start_dt"] <= now:
|
|
return "active"
|
|
return "scheduled"
|
|
|
|
|
|
@router.get("")
|
|
async def list_windows(site_id: str = "sg-01"):
|
|
return [
|
|
{**w, "status": _window_status(w)}
|
|
for w in _windows
|
|
if w["site_id"] == site_id
|
|
]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_window(body: WindowCreate):
|
|
window = {
|
|
"id": str(uuid.uuid4())[:8],
|
|
"site_id": body.site_id,
|
|
"title": body.title,
|
|
"target": body.target,
|
|
"target_label": body.target_label,
|
|
"start_dt": body.start_dt,
|
|
"end_dt": body.end_dt,
|
|
"suppress_alarms": body.suppress_alarms,
|
|
"notes": body.notes,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
_windows.append(window)
|
|
return {**window, "status": _window_status(window)}
|
|
|
|
|
|
@router.delete("/{window_id}", status_code=204)
|
|
async def delete_window(window_id: str):
|
|
global _windows
|
|
before = len(_windows)
|
|
_windows = [w for w in _windows if w["id"] != window_id]
|
|
if len(_windows) == before:
|
|
raise HTTPException(status_code=404, detail="Window not found")
|
|
|
|
|
|
@router.get("/active")
|
|
async def active_windows(site_id: str = "sg-01"):
|
|
"""Returns only currently active windows — used by alarm page for suppression check."""
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
return [
|
|
w for w in _windows
|
|
if w["site_id"] == site_id
|
|
and w["start_dt"] <= now <= w["end_dt"]
|
|
and w["suppress_alarms"]
|
|
]
|