alro65
cfd94f905a
- Restrict CORS to localhost origins (was allow_origins=[*])
- Require valid JWT on WebSocket /ws (anonymous no longer gets admin view)
- Fix path traversal in delete_cell(): resolve() + parent check
- Validate cell_id format in /charts/download-noaa/{cell_id}
- Exclude charts/ and Cartas/ from git (keep US1GC09M world overview)
- Add NOAA ENC Portal external link in charts catalog tab
- Untrack __pycache__/, .db, .claude/ session files
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""
|
|
CRUD for the lamp catalog. Each Aid references a lamp by id; the lamp's
|
|
voltage_min / voltage_max determine that aid's battery warning/alarm
|
|
thresholds (see compute_thresholds).
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
import uuid
|
|
|
|
from database import get_db
|
|
from models.lamp import Lamp
|
|
from models.aid import Aid
|
|
|
|
router = APIRouter(prefix="/lamps", tags=["lamps"])
|
|
|
|
|
|
class LampIn(BaseModel):
|
|
manufacturer: str
|
|
model: str
|
|
lamp_count: int = 1
|
|
voltage_min: float = Field(..., gt=0)
|
|
voltage_max: float = Field(..., gt=0)
|
|
warn_pct: float = Field(20.0, ge=1, le=50) # % of range → warning
|
|
alarm_pct: float = Field(10.0, ge=1, le=50) # % of range → alarm
|
|
notes: Optional[str] = None
|
|
|
|
|
|
def compute_thresholds(vmin: float, vmax: float,
|
|
warn_pct: float = 20.0, alarm_pct: float = 10.0) -> dict:
|
|
rng = vmax - vmin
|
|
return {
|
|
"warn_v": round(vmin + rng * (warn_pct / 100), 3),
|
|
"alarm_v": round(vmin + rng * (alarm_pct / 100), 3),
|
|
}
|
|
|
|
|
|
def _lamp_dict(l: Lamp) -> dict:
|
|
th = compute_thresholds(l.voltage_min, l.voltage_max,
|
|
l.warn_pct or 20.0, l.alarm_pct or 10.0)
|
|
return {
|
|
"id": l.id,
|
|
"manufacturer": l.manufacturer,
|
|
"model": l.model,
|
|
"lamp_count": l.lamp_count,
|
|
"voltage_min": l.voltage_min,
|
|
"voltage_max": l.voltage_max,
|
|
"warn_pct": l.warn_pct or 20.0,
|
|
"alarm_pct": l.alarm_pct or 10.0,
|
|
"notes": l.notes,
|
|
"warn_v": th["warn_v"],
|
|
"alarm_v": th["alarm_v"],
|
|
}
|
|
|
|
|
|
@router.get("/")
|
|
def list_lamps(db: Session = Depends(get_db)):
|
|
return [_lamp_dict(l) for l in db.query(Lamp).order_by(Lamp.manufacturer, Lamp.model).all()]
|
|
|
|
|
|
@router.post("/")
|
|
def create_lamp(data: LampIn, db: Session = Depends(get_db)):
|
|
if data.voltage_max <= data.voltage_min:
|
|
raise HTTPException(400, "voltage_max must be greater than voltage_min")
|
|
lamp = Lamp(id=str(uuid.uuid4()), **data.model_dump())
|
|
db.add(lamp)
|
|
db.commit(); db.refresh(lamp)
|
|
return _lamp_dict(lamp)
|
|
|
|
|
|
@router.put("/{lamp_id}")
|
|
def update_lamp(lamp_id: str, data: LampIn, db: Session = Depends(get_db)):
|
|
lamp = db.query(Lamp).filter(Lamp.id == lamp_id).first()
|
|
if not lamp:
|
|
raise HTTPException(404, "Lamp not found")
|
|
if data.voltage_max <= data.voltage_min:
|
|
raise HTTPException(400, "voltage_max must be greater than voltage_min")
|
|
for k, v in data.model_dump().items():
|
|
setattr(lamp, k, v)
|
|
db.commit(); db.refresh(lamp)
|
|
return _lamp_dict(lamp)
|
|
|
|
|
|
@router.delete("/{lamp_id}")
|
|
def delete_lamp(lamp_id: str, force: bool = False, db: Session = Depends(get_db)):
|
|
in_use = db.query(Aid).filter(Aid.lamp_id == lamp_id).count()
|
|
if in_use and not force:
|
|
# Frontend will retry with force=true after user confirms
|
|
raise HTTPException(409, f"Lamp is in use by {in_use} aid(s)")
|
|
if in_use and force:
|
|
# Unassign all aids that point at this lamp
|
|
db.query(Aid).filter(Aid.lamp_id == lamp_id).update({"lamp_id": None})
|
|
db.query(Lamp).filter(Lamp.id == lamp_id).delete()
|
|
db.commit()
|
|
return {"deleted": lamp_id, "unassigned_from": in_use}
|