gotthard_g/aaredb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added dewar type, serial number, generate unique id, qr code and generate label

Browse Source
This commit is contained in:
GotthardG
2024-11-14 23:17:20 +01:00
parent ca11a359f9
commit 6083c72a1d
8 changed files with 684 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
backend/app/routers/dewar.py
View File

@ -1,26 +1,50 @@
from fastapi import APIRouter, HTTPException, status, Depends
from fastapi import APIRouter, HTTPException, status, Depends, Response
from sqlalchemy.orm import Session, joinedload
from typing import List
import logging
from sqlalchemy.exc import SQLAlchemyError
from pydantic import ValidationError
from app.schemas import Dewar as DewarSchema, DewarCreate, DewarUpdate
from app.models import Dewar as DewarModel, Puck as PuckModel, \
Sample as SampleModel # Assuming SampleModel is defined in models
from app.schemas import (
Dewar as DewarSchema,
DewarCreate,
DewarUpdate,
DewarType as DewarTypeSchema,
DewarTypeCreate,
DewarSerialNumber as DewarSerialNumberSchema,
DewarSerialNumberCreate
)
from app.models import (
Dewar as DewarModel,
Puck as PuckModel,
Sample as SampleModel,
DewarType as DewarTypeModel,
DewarSerialNumber as DewarSerialNumberModel
)
from app.dependencies import get_db
import uuid
import qrcode
import io
from io import BytesIO
from PIL import Image
from reportlab.lib.pagesizes import A5
from reportlab.lib.units import cm
from reportlab.pdfgen import canvas
router = APIRouter()
@router.get("/", response_model=List[DewarSchema])
async def get_dewars(db: Session = Depends(get_db)):
dewars = db.query(DewarModel).options(joinedload(DewarModel.pucks)).all()
return dewars
def generate_unique_id(db: Session) -> str:
while True:
unique_id = str(uuid.uuid4())
existing_dewar = db.query(DewarModel).filter(DewarModel.unique_id == unique_id).first()
if not existing_dewar:
break
return unique_id
@router.post("/", response_model=DewarSchema, status_code=status.HTTP_201_CREATED)
async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> DewarSchema:
try:
unique_id = generate_unique_id(db)
db_dewar = DewarModel(
dewar_name=dewar.dewar_name,
tracking_number=dewar.tracking_number,
@ -29,11 +53,10 @@ async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> Dew
shipping_date=dewar.shipping_date,
arrival_date=dewar.arrival_date,
returning_date=dewar.returning_date,
qrcode=dewar.qrcode,
contact_person_id=dewar.contact_person_id,
return_address_id=dewar.return_address_id
return_address_id=dewar.return_address_id,
unique_id=unique_id
)
db.add(db_dewar)
db.commit()
db.refresh(db_dewar)
@ -54,7 +77,6 @@ async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> Dew
puck_id=puck.id,
sample_name=sample_data.sample_name,
position=sample_data.position,
# Ensure only valid attributes are set
data_collection_parameters=sample_data.data_collection_parameters,
)
db.add(sample)
@ -70,24 +92,142 @@ async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> Dew
logging.error(f"Validation error occurred: {e}")
raise HTTPException(status_code=400, detail="Validation error")
@router.post("/{dewar_id}/generate-qrcode")
async def generate_dewar_qrcode(dewar_id: int, db: Session = Depends(get_db)):
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
if not dewar:
raise HTTPException(status_code=404, detail="Dewar not found")
if not dewar.unique_id:
dewar.unique_id = generate_unique_id(db)
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(dewar.unique_id)
qr.make(fit=True)
img = qr.make_image(fill='black', back_color='white')
buf = io.BytesIO()
img.save(buf)
buf.seek(0)
dewar.qrcode = dewar.unique_id
dewar.qrcode_image = buf.getvalue()
db.commit()
return {"message": "QR Code generated", "qrcode": dewar.unique_id}
def generate_label(dewar):
buffer = BytesIO()
c = canvas.Canvas(buffer, pagesize=A5)
# Draw header
c.setFont("Helvetica-Bold", 16)
c.drawCentredString(10.5 * cm, 14 * cm, "COMPANY LOGO / TITLE")
# Draw details section
c.setFont("Helvetica", 12)
c.drawString(2 * cm, 12.5 * cm, f"Dewar Name: {dewar.dewar_name}")
c.drawString(2 * cm, 11.5 * cm, f"Unique ID: {dewar.unique_id}")
if dewar.dewar_type:
c.drawString(2 * cm, 10.5 * cm, f"Dewar Type: {dewar.dewar_type.dewar_type}")
else:
c.drawString(2 * cm, 10.5 * cm, "Dewar Type: Unknown")
c.drawString(2 * cm, 9.5 * cm, "Beamtime Information: Placeholder")
# Generate QR code
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(dewar.unique_id)
qr.make(fit=True)
img = qr.make_image(fill='black', back_color='white')
qr_io = BytesIO()
img.save(qr_io, format='PNG')
qr_io.seek(0)
qr_image = Image.open(qr_io)
# Add QR code to PDF
c.drawInlineImage(qr_image, 8 * cm, 5 * cm, width=4 * cm, height=4 * cm)
# Add footer text
c.setFont("Helvetica", 10)
c.drawCentredString(10.5 * cm, 4 * cm, "Scan for more information")
# Draw border
c.rect(1 * cm, 3 * cm, 18 * cm, 12 * cm)
# Finalize the canvas
c.showPage()
c.save()
buffer.seek(0)
return buffer
@router.get("/{dewar_id}/download-label", response_class=Response)
async def download_dewar_label(dewar_id: int, db: Session = Depends(get_db)):
dewar = db.query(DewarModel).options(joinedload(DewarModel.dewar_type)).filter(DewarModel.id == dewar_id).first()
if not dewar:
raise HTTPException(status_code=404, detail="Dewar not found")
if not dewar.unique_id:
raise HTTPException(status_code=404, detail="QR Code not generated for this dewar")
buffer = generate_label(dewar)
return Response(buffer.getvalue(), media_type="application/pdf", headers={
"Content-Disposition": f"attachment; filename=dewar_label_{dewar.id}.pdf"
})
@router.get("/", response_model=List[DewarSchema])
async def get_dewars(db: Session = Depends(get_db)):
try:
dewars = db.query(DewarModel).options(joinedload(DewarModel.pucks)).all()
return dewars
except SQLAlchemyError as e:
logging.error(f"Database error occurred: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/dewar-types", response_model=List[DewarTypeSchema])
def get_dewar_types(db: Session = Depends(get_db)):
return db.query(DewarTypeModel).all()
@router.get("/dewar-types/{type_id}/serial-numbers", response_model=List[DewarSerialNumberSchema])
def get_serial_numbers(type_id: int, db: Session = Depends(get_db)):
return db.query(DewarSerialNumberModel).filter(DewarSerialNumberModel.dewar_type_id == type_id).all()
@router.post("/dewar-types", response_model=DewarTypeSchema)
def create_dewar_type(dewar_type: DewarTypeCreate, db: Session = Depends(get_db)):
db_type = DewarTypeModel(**dewar_type.dict())
db.add(db_type)
db.commit()
db.refresh(db_type)
return db_type
@router.post("/dewar-serial-numbers", response_model=DewarSerialNumberSchema)
def create_dewar_serial_number(serial_number: DewarSerialNumberCreate, db: Session = Depends(get_db)):
db_serial = DewarSerialNumberModel(**serial_number.dict())
db.add(db_serial)
db.commit()
db.refresh(db_serial)
return db_serial
@router.get("/dewar-serial-numbers", response_model=List[DewarSerialNumberSchema])
def get_all_serial_numbers(db: Session = Depends(get_db)):
try:
serial_numbers = db.query(DewarSerialNumberModel).all()
return serial_numbers
except SQLAlchemyError as e:
logging.error(f"Database error occurred: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/{dewar_id}", response_model=DewarSchema)
async def get_dewar(dewar_id: int, db: Session = Depends(get_db)):
dewar = db.query(DewarModel).options(
joinedload(DewarModel.pucks).joinedload(PuckModel.positions)
joinedload(DewarModel.pucks).joinedload(PuckModel.samples)
).filter(DewarModel.id == dewar_id).first()
if not dewar:
raise HTTPException(status_code=404, detail="Dewar not found")
# Ensure dewar.pucks is an empty list if there are no pucks
dewar_dict = dewar.__dict__
if dewar_dict.get("pucks") is None:
dewar_dict["pucks"] = []
return DewarSchema.from_orm(dewar)
@router.put("/{dewar_id}", response_model=DewarSchema)
async def update_dewar(dewar_id: int, dewar_update: DewarUpdate, db: Session = Depends(get_db)) -> DewarSchema:
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
@ -96,22 +236,20 @@ async def update_dewar(dewar_id: int, dewar_update: DewarUpdate, db: Session = D
raise HTTPException(status_code=404, detail="Dewar not found")
for key, value in dewar_update.dict(exclude_unset=True).items():
# Ensure we're only setting directly settable attributes
if hasattr(dewar, key):
setattr(dewar, key, value)
db.commit()
db.refresh(dewar)
return dewar
@router.delete("/{dewar_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_dewar(dewar_id: int, db: Session = Depends(get_db)):
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
if not dewar:
raise HTTPException(status_code=404, detail="Dewar not found")
db.delete(dewar)
db.commit()
return
return