113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
from fastapi import APIRouter, HTTPException, Depends
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime, timedelta
|
|
from typing import List
|
|
from app.models import Dewar as DewarModel, Slot as SlotModel, LogisticsEvent as LogisticsEventModel
|
|
from app.schemas import LogisticsEventCreate, Slot as SlotSchema, Dewar as DewarSchema
|
|
from app.database import get_db
|
|
from app.data import slots_data
|
|
import logging
|
|
|
|
router = APIRouter()
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.get("/dewars", response_model=List[DewarSchema])
|
|
async def get_all_dewars(db: Session = Depends(get_db)):
|
|
dewars = db.query(DewarModel).all()
|
|
return dewars
|
|
|
|
|
|
@router.get("/dewar/{unique_id}", response_model=DewarSchema)
|
|
async def get_dewar_by_unique_id(unique_id: str, db: Session = Depends(get_db)):
|
|
dewar = db.query(DewarModel).filter(DewarModel.unique_id == unique_id.strip()).first()
|
|
if not dewar:
|
|
raise HTTPException(status_code=404, detail="Dewar not found")
|
|
return dewar
|
|
|
|
|
|
@router.post("/dewar/scan", response_model=dict)
|
|
async def scan_dewar(event_data: LogisticsEventCreate, db: Session = Depends(get_db)):
|
|
dewar_qr_code = event_data.dewar_qr_code
|
|
location_qr_code = event_data.location_qr_code
|
|
transaction_type = event_data.transaction_type
|
|
|
|
dewar = db.query(DewarModel).filter(DewarModel.unique_id == dewar_qr_code).first()
|
|
if not dewar:
|
|
raise HTTPException(status_code=404, detail="Dewar not found")
|
|
|
|
slot = db.query(SlotModel).filter(SlotModel.qr_code == location_qr_code).first()
|
|
if transaction_type == 'incoming':
|
|
if not slot or slot.occupied:
|
|
raise HTTPException(status_code=400, detail="Slot not found or already occupied")
|
|
slot.occupied = True
|
|
|
|
elif transaction_type == 'outgoing':
|
|
if not slot or not slot.occupied:
|
|
raise HTTPException(status_code=400, detail="Slot not found or not occupied")
|
|
slot.occupied = False
|
|
|
|
# Log the event
|
|
log_event(db, dewar.id, slot.id if slot else None, transaction_type)
|
|
|
|
db.commit()
|
|
return {"message": "Status updated successfully"}
|
|
|
|
|
|
def log_event(db: Session, dewar_id: int, slot_id: str, event_type: str):
|
|
new_event = LogisticsEventModel(dewar_id=dewar_id, slot_id=slot_id, event_type=event_type)
|
|
db.add(new_event)
|
|
db.commit()
|
|
|
|
|
|
# Convert SQLAlchemy model to dictionary (if necessary)
|
|
def slot_to_dict(slot: SlotModel) -> dict:
|
|
return {
|
|
"id": slot.id,
|
|
"qr_code": slot.qr_code,
|
|
"label": slot.label,
|
|
"qr_base": slot.qr_base,
|
|
"occupied": slot.occupied,
|
|
"needs_refill": slot.needs_refill,
|
|
"last_refill": slot.last_refill.isoformat(),
|
|
"time_until_refill": str(slot.time_until_refill) if slot.time_until_refill else None # Ensure correct format
|
|
}
|
|
|
|
@router.get("/slots", response_model=List[dict])
|
|
def read_slots(db: Session = Depends(get_db)):
|
|
return [slot_to_dict(slot) for slot in db.query(SlotModel).all()]
|
|
|
|
|
|
@router.get("/dewars/refill-status", response_model=List[DewarSchema])
|
|
async def refill_status(db: Session = Depends(get_db)):
|
|
dewars = db.query(DewarModel).all()
|
|
result = []
|
|
current_time = datetime.utcnow()
|
|
|
|
for dewar in dewars:
|
|
last_refill_event = (
|
|
db.query(LogisticsEventModel)
|
|
.filter(
|
|
LogisticsEventModel.dewar_id == dewar.id,
|
|
LogisticsEventModel.event_type == 'refill'
|
|
)
|
|
.order_by(LogisticsEventModel.timestamp.desc())
|
|
.first()
|
|
)
|
|
|
|
time_until_refill = None
|
|
if last_refill_event:
|
|
time_until_refill = last_refill_event.timestamp + timedelta(hours=24) - current_time
|
|
|
|
dewar_data = DewarSchema(
|
|
id=dewar.id,
|
|
dewar_name=dewar.dewar_name,
|
|
unique_id=dewar.unique_id,
|
|
time_until_refill=str(time_until_refill) if time_until_refill else None # Ensure correct format
|
|
)
|
|
result.append(dewar_data)
|
|
|
|
return result |