Coverage for app\routers\calibration.py: 32%

1from typing import List 

2from fastapi import APIRouter, Depends, HTTPException 

3from sqlalchemy.orm import Session 

4from ..db import SessionLocal, CalibrationZone, CameraBinding, School 

5from ..schemas import calibration as schemas 

6from ..deps import get_db 

7 

8router = APIRouter() 

9 

10# Zones 

11@router.post("/zones", response_model=schemas.CalibrationZone) 

12def create_zone(zone: schemas.CalibrationZoneCreate, db: Session = Depends(get_db)): 

13 # Verify school exists 

14 school = db.query(School).filter(School.id == zone.school_id).first() 

15 if not school: 

16 raise HTTPException(status_code=404, detail="School not found") 

17 

18 db_zone = CalibrationZone( 

19 school_id=zone.school_id, 

20 name=zone.name, 

21 type=zone.type, 

22 points=[p.model_dump() for p in zone.points], 

23 metadata_json=zone.metadata_json 

24 ) 

25 db.add(db_zone) 

26 db.commit() 

27 db.refresh(db_zone) 

28 return db_zone 

29 

30@router.get("/zones/school/{school_id}", response_model=List[schemas.CalibrationZone]) 

31def list_zones_by_school(school_id: str, db: Session = Depends(get_db)): 

32 return db.query(CalibrationZone).filter(CalibrationZone.school_id == school_id).all() 

33 

34@router.get("/zones/{zone_id}", response_model=schemas.CalibrationZone) 

35def get_zone(zone_id: str, db: Session = Depends(get_db)): 

36 zone = db.query(CalibrationZone).filter(CalibrationZone.id == zone_id).first() 

37 if not zone: 

38 raise HTTPException(status_code=404, detail="Zone not found") 

39 return zone 

40 

41@router.put("/zones/{zone_id}", response_model=schemas.CalibrationZone) 

42def update_zone(zone_id: str, zone_update: schemas.CalibrationZoneUpdate, db: Session = Depends(get_db)): 

43 db_zone = db.query(CalibrationZone).filter(CalibrationZone.id == zone_id).first() 

44 if not db_zone: 

45 raise HTTPException(status_code=404, detail="Zone not found") 

46 

47 if zone_update.name is not None: 

48 db_zone.name = zone_update.name 

49 if zone_update.type is not None: 

50 db_zone.type = zone_update.type 

51 if zone_update.points is not None: 

52 db_zone.points = [p.model_dump() for p in zone_update.points] 

53 if zone_update.metadata_json is not None: 

54 db_zone.metadata_json = zone_update.metadata_json 

55 

56 db.commit() 

57 db.refresh(db_zone) 

58 return db_zone 

59 

60@router.delete("/zones/{zone_id}") 

61def delete_zone(zone_id: str, db: Session = Depends(get_db)): 

62 db_zone = db.query(CalibrationZone).filter(CalibrationZone.id == zone_id).first() 

63 if not db_zone: 

64 raise HTTPException(status_code=404, detail="Zone not found") 

65 

66 db.delete(db_zone) 

67 db.commit() 

68 return {"status": "deleted"} 

69 

70# Camera Bindings 

71@router.post("/bindings", response_model=schemas.CameraBinding) 

72def create_binding(binding: schemas.CameraBindingCreate, db: Session = Depends(get_db)): 

73 # Verify zone exists 

74 zone = db.query(CalibrationZone).filter(CalibrationZone.id == binding.zone_id).first() 

75 if not zone: 

76 raise HTTPException(status_code=404, detail="Zone not found") 

77 

78 # Fix common typo 

79 if binding.stream_url and binding.stream_url.startswith("rstp://"): 

80 binding.stream_url = binding.stream_url.replace("rstp://", "rtsp://", 1) 

81 

82 db_binding = CameraBinding( 

83 zone_id=binding.zone_id, 

84 camera_id=binding.camera_id, 

85 stream_url=binding.stream_url, 

86 status=binding.status 

87 ) 

88 db.add(db_binding) 

89 db.commit() 

90 db.refresh(db_binding) 

91 return db_binding 

92 

93@router.get("/bindings/zone/{zone_id}", response_model=List[schemas.CameraBinding]) 

94def list_bindings_by_zone(zone_id: str, db: Session = Depends(get_db)): 

95 return db.query(CameraBinding).filter(CameraBinding.zone_id == zone_id).all() 

96 

97@router.delete("/bindings/{binding_id}") 

98def delete_binding(binding_id: str, db: Session = Depends(get_db)): 

99 db_binding = db.query(CameraBinding).filter(CameraBinding.id == binding_id).first() 

100 if not db_binding: 

101 raise HTTPException(status_code=404, detail="Binding not found") 

102 

103 db.delete(db_binding) 

104 db.commit() 

105 return {"status": "deleted"}