9739b8cfe9
Updated the `number_of_pucks` and `number_of_samples` fields in the `schemas.py` to be optional for greater flexibility. Simplified the test Jupyter Notebook by restructuring imports and consolidating function calls for better readability and maintainability.
677 lines
19 KiB
Python
677 lines
19 KiB
Python
from typing import List, Optional
|
|
from datetime import datetime
|
|
from pydantic import BaseModel, EmailStr, constr, Field, field_validator
|
|
from datetime import date
|
|
import logging
|
|
import re
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class loginToken(BaseModel):
|
|
access_token: str
|
|
token_type: str
|
|
|
|
|
|
class loginData(BaseModel):
|
|
username: str
|
|
pgroups: List[int]
|
|
|
|
|
|
class DewarTypeBase(BaseModel):
|
|
dewar_type: str
|
|
|
|
|
|
class DewarTypeCreate(DewarTypeBase):
|
|
pass
|
|
|
|
|
|
class DewarType(DewarTypeBase):
|
|
id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class DewarSerialNumberBase(BaseModel):
|
|
serial_number: str
|
|
dewar_type_id: int
|
|
|
|
|
|
class DewarSerialNumberCreate(DewarSerialNumberBase):
|
|
pass
|
|
|
|
|
|
class DewarSerialNumber(DewarSerialNumberBase):
|
|
id: int
|
|
dewar_type: DewarType
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class DataCollectionParameters(BaseModel):
|
|
directory: Optional[str] = None
|
|
oscillation: Optional[float] = None # Only accept positive float
|
|
exposure: Optional[float] = None # Only accept positive floats between 0 and 1
|
|
totalrange: Optional[int] = None # Only accept positive integers between 0 and 360
|
|
transmission: Optional[
|
|
int
|
|
] = None # Only accept positive integers between 0 and 100
|
|
targetresolution: Optional[float] = None # Only accept positive float
|
|
aperture: Optional[str] = None # Optional string field
|
|
datacollectiontype: Optional[
|
|
str
|
|
] = None # Only accept "standard", other types might be added later
|
|
processingpipeline: Optional[
|
|
str
|
|
] = "" # Only accept "gopy", "autoproc", "xia2dials"
|
|
spacegroupnumber: Optional[
|
|
int
|
|
] = None # Only accept positive integers between 1 and 230
|
|
cellparameters: Optional[
|
|
str
|
|
] = None # Must be a set of six positive floats or integers
|
|
rescutkey: Optional[str] = None # Only accept "is" or "cchalf"
|
|
rescutvalue: Optional[
|
|
float
|
|
] = None # Must be a positive float if rescutkey is provided
|
|
userresolution: Optional[float] = None
|
|
pdbid: Optional[
|
|
str
|
|
] = "" # Accepts either the format of the protein data bank code or {provided}
|
|
autoprocfull: Optional[bool] = None
|
|
procfull: Optional[bool] = None
|
|
adpenabled: Optional[bool] = None
|
|
noano: Optional[bool] = None
|
|
ffcscampaign: Optional[bool] = None
|
|
trustedhigh: Optional[float] = None # Should be a float between 0 and 2.0
|
|
autoprocextraparams: Optional[str] = None # Optional string field
|
|
chiphiangles: Optional[float] = None # Optional float field between 0 and 30
|
|
dose: Optional[float] = None # Optional float field
|
|
|
|
def to_dict(self):
|
|
"""Convert the model instance to a dictionary."""
|
|
return self.dict(
|
|
exclude_unset=True
|
|
) # Use this built-in method for serialization
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
@field_validator("directory", mode="after")
|
|
@classmethod
|
|
def directory_characters(cls, v):
|
|
logger.debug(f"Validating 'directory' field with initial value: {repr(v)}")
|
|
|
|
# Default directory value if empty
|
|
if not v: # Handles None or empty cases
|
|
default_value = "{sgPuck}/{sgPosition}"
|
|
logger.warning(
|
|
f"'directory' field is empty or None. Assigning default value: "
|
|
f"{default_value}"
|
|
)
|
|
return default_value
|
|
|
|
# Strip trailing slashes and store original value for comparison
|
|
v = str(v).strip("/") # Ensure it's a string and no trailing slashes
|
|
original_value = v
|
|
|
|
# Replace spaces with underscores
|
|
v = v.replace(" ", "_")
|
|
logger.debug(f"Corrected 'directory', spaces replaced: {repr(v)}")
|
|
|
|
# Validate directory pattern with macros and allowed characters
|
|
valid_macros = [
|
|
"{date}",
|
|
"{prefix}",
|
|
"{sgPuck}",
|
|
"{sgPosition}",
|
|
"{beamline}",
|
|
"{sgPrefix}",
|
|
"{sgPriority}",
|
|
"{protein}",
|
|
"{method}",
|
|
]
|
|
valid_macro_pattern = re.compile(
|
|
"|".join(re.escape(macro) for macro in valid_macros)
|
|
)
|
|
|
|
# Check if the value contains valid macros
|
|
allowed_chars_pattern = "[a-z0-9_.+-/]"
|
|
v_without_macros = valid_macro_pattern.sub("macro", v)
|
|
|
|
allowed_path_pattern = re.compile(
|
|
f"^(({allowed_chars_pattern}+|macro)*/*)*$", re.IGNORECASE
|
|
)
|
|
if not allowed_path_pattern.match(v_without_macros):
|
|
raise ValueError(
|
|
f"'{v}' is not valid. Value must be a valid path or macro."
|
|
)
|
|
|
|
# Log and return corrected value
|
|
if v != original_value:
|
|
logger.info(f"Directory was corrected from '{original_value}' to '{v}'")
|
|
return v
|
|
|
|
@field_validator("aperture", mode="before")
|
|
@classmethod
|
|
def aperture_selection(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = int(float(v))
|
|
if v not in {1, 2, 3}:
|
|
raise ValueError(f" '{v}' is not valid. Value must be 1, 2, or 3.")
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be 1, 2, or 3."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("oscillation", mode="before")
|
|
@classmethod
|
|
def positive_float_validator(cls, v):
|
|
if v is None:
|
|
return None
|
|
try:
|
|
v = float(v)
|
|
if v <= 0:
|
|
raise ValueError(f"'{v}' is not valid. Value must be a positive float.")
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f"'{v}' is not valid. Value must be a positive float."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("exposure", mode="before")
|
|
@classmethod
|
|
def exposure_in_range(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = float(v)
|
|
if not (0 <= v <= 1):
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be a float between 0 and 1."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be a float between 0 and 1."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("totalrange", mode="before")
|
|
@classmethod
|
|
def totalrange_in_range(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = int(v)
|
|
if not (0 <= v <= 360):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 0 and 360."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 0 and 360."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("transmission", mode="before")
|
|
@classmethod
|
|
def transmission_fraction(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = int(v)
|
|
if not (0 <= v <= 100):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 0 and 100."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 0 and 100."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("datacollectiontype", mode="before")
|
|
@classmethod
|
|
def datacollectiontype_allowed(cls, v):
|
|
allowed = {"standard"} # Other types of data collection might be added later
|
|
if v and v.lower() not in allowed:
|
|
raise ValueError(f" '{v}' is not valid." f"Value must be one of {allowed}.")
|
|
return v
|
|
|
|
@field_validator("processingpipeline", mode="before")
|
|
@classmethod
|
|
def processingpipeline_allowed(cls, v):
|
|
allowed = {"gopy", "autoproc", "xia2dials"}
|
|
if v and v.lower() not in allowed:
|
|
raise ValueError(f" '{v}' is not valid." f"Value must be one of {allowed}.")
|
|
return v
|
|
|
|
@field_validator("spacegroupnumber", mode="before")
|
|
@classmethod
|
|
def spacegroupnumber_allowed(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = int(v)
|
|
if not (1 <= v <= 230):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 1 and 230."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be an integer between 1 and 230."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("cellparameters", mode="before")
|
|
@classmethod
|
|
def cellparameters_format(cls, v):
|
|
if v:
|
|
values = [float(i) for i in v.split(",")]
|
|
if len(values) != 6 or any(val <= 0 for val in values):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be a set of six positive floats or integers."
|
|
)
|
|
return v
|
|
|
|
# @field_validator("rescutkey", "rescutvalue", mode="before")
|
|
# @classmethod
|
|
# def rescutkey_value_pair(cls, values):
|
|
# rescutkey = values.get("rescutkey")
|
|
# rescutvalue = values.get("rescutvalue")
|
|
# if rescutkey and rescutvalue:
|
|
# if rescutkey not in {"is", "cchalf"}:
|
|
# raise ValueError("Rescutkey must be either 'is' or 'cchalf'")
|
|
# if not isinstance(rescutvalue, float) or rescutvalue <= 0:
|
|
# raise ValueError(
|
|
# "Rescutvalue must be a positive float if rescutkey is provided"
|
|
# )
|
|
# return values
|
|
|
|
@field_validator("trustedhigh", mode="before")
|
|
@classmethod
|
|
def trustedhigh_allowed(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = float(v)
|
|
if not (0 <= v <= 2.0):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be a float between 0 and 2.0."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid." f"Value must be a float between 0 and 2.0."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("chiphiangles", mode="before")
|
|
@classmethod
|
|
def chiphiangles_allowed(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = float(v)
|
|
if not (0 <= v <= 30):
|
|
raise ValueError(
|
|
f" '{v}' is not valid."
|
|
f"Value must be a float between 0 and 30."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be a float between 0 and 30."
|
|
) from e
|
|
return v
|
|
|
|
@field_validator("dose", mode="before")
|
|
@classmethod
|
|
def dose_positive(cls, v):
|
|
if v is not None:
|
|
try:
|
|
v = float(v)
|
|
if v <= 0:
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be a positive float."
|
|
)
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError(
|
|
f" '{v}' is not valid. Value must be a positive float."
|
|
) from e
|
|
return v
|
|
|
|
|
|
class SampleEventCreate(BaseModel):
|
|
event_type: str
|
|
|
|
|
|
class SampleEventResponse(BaseModel):
|
|
id: int
|
|
sample_id: int
|
|
event_type: str
|
|
timestamp: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class Results(BaseModel):
|
|
# Define attributes for Results here
|
|
pass
|
|
|
|
|
|
class ContactPersonBase(BaseModel):
|
|
firstname: str
|
|
lastname: str
|
|
phone_number: str
|
|
email: EmailStr
|
|
|
|
|
|
class ContactPersonCreate(ContactPersonBase):
|
|
pass
|
|
|
|
|
|
class ContactPerson(ContactPersonBase):
|
|
id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class ContactPersonUpdate(BaseModel):
|
|
firstname: Optional[str] = None
|
|
lastname: Optional[str] = None
|
|
phone_number: Optional[str] = None
|
|
email: Optional[EmailStr] = None
|
|
|
|
|
|
class AddressCreate(BaseModel):
|
|
street: str
|
|
city: str
|
|
zipcode: str
|
|
country: str
|
|
|
|
|
|
class Address(AddressCreate):
|
|
id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AddressUpdate(BaseModel):
|
|
street: Optional[str] = None
|
|
city: Optional[str] = None
|
|
zipcode: Optional[str] = None
|
|
country: Optional[str] = None
|
|
|
|
|
|
class Sample(BaseModel):
|
|
id: int
|
|
sample_name: str
|
|
position: int # Position within the puck
|
|
puck_id: int
|
|
crystalname: Optional[str] = Field(None)
|
|
proteinname: Optional[str] = None
|
|
positioninpuck: Optional[int] = Field(None)
|
|
priority: Optional[int] = None
|
|
comments: Optional[str] = None
|
|
data_collection_parameters: Optional[DataCollectionParameters]
|
|
events: List[SampleEventResponse] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class SampleCreate(BaseModel):
|
|
sample_name: str = Field(..., alias="crystalname")
|
|
proteinname: Optional[str] = None
|
|
position: int = Field(..., alias="positioninpuck")
|
|
data_collection_parameters: Optional[DataCollectionParameters] = Field(default=None)
|
|
priority: Optional[int] = None
|
|
comments: Optional[str] = None
|
|
results: Optional[Results] = None
|
|
events: Optional[List[str]] = None
|
|
|
|
class Config:
|
|
populate_by_name = True
|
|
|
|
|
|
class PuckEvent(BaseModel):
|
|
id: int
|
|
puck_id: int
|
|
tell_position: Optional[str] = None
|
|
event_type: str
|
|
timestamp: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class PuckBase(BaseModel):
|
|
puck_name: str
|
|
puck_type: str
|
|
puck_location_in_dewar: int
|
|
|
|
|
|
class PuckCreate(BaseModel):
|
|
puck_name: str
|
|
puck_type: str
|
|
puck_location_in_dewar: int
|
|
samples: List[SampleCreate] = []
|
|
|
|
|
|
class PuckUpdate(BaseModel):
|
|
puck_name: Optional[str] = None
|
|
puck_type: Optional[str] = None
|
|
puck_location_in_dewar: Optional[int] = None
|
|
dewar_id: Optional[int] = None
|
|
|
|
|
|
class Puck(BaseModel):
|
|
id: int
|
|
puck_name: str
|
|
puck_type: str
|
|
puck_location_in_dewar: int
|
|
dewar_id: int
|
|
events: List[PuckEvent] = []
|
|
samples: List[Sample] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class DewarBase(BaseModel):
|
|
dewar_name: str
|
|
dewar_type_id: Optional[int] = None
|
|
dewar_serial_number_id: Optional[int] = None
|
|
unique_id: Optional[str] = None
|
|
tracking_number: str
|
|
number_of_pucks: Optional[int] = None
|
|
number_of_samples: Optional[int] = None
|
|
status: str
|
|
ready_date: Optional[date]
|
|
shipping_date: Optional[date]
|
|
arrival_date: Optional[date]
|
|
returning_date: Optional[date]
|
|
contact_person_id: Optional[int]
|
|
return_address_id: Optional[int]
|
|
pucks: List[PuckCreate] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class DewarCreate(DewarBase):
|
|
pass
|
|
|
|
|
|
class Dewar(DewarBase):
|
|
id: int
|
|
shipment_id: Optional[int]
|
|
contact_person: Optional[ContactPerson]
|
|
return_address: Optional[Address]
|
|
pucks: List[Puck] = [] # List of pucks within this dewar
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class DewarUpdate(BaseModel):
|
|
dewar_name: Optional[str] = None
|
|
dewar_type_id: Optional[int] = None
|
|
dewar_serial_number_id: Optional[int] = None
|
|
unique_id: Optional[str] = None
|
|
tracking_number: Optional[str] = None
|
|
status: Optional[str] = None
|
|
ready_date: Optional[date] = None
|
|
shipping_date: Optional[date] = None
|
|
arrival_date: Optional[date] = None
|
|
returning_date: Optional[date] = None
|
|
contact_person_id: Optional[int] = None
|
|
address_id: Optional[int] = None
|
|
|
|
|
|
class DewarSchema(BaseModel):
|
|
id: int
|
|
dewar_name: str
|
|
tracking_number: str
|
|
status: str
|
|
contact_person_id: int
|
|
return_address_id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class Proposal(BaseModel):
|
|
id: int
|
|
number: str
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class Shipment(BaseModel):
|
|
id: int
|
|
shipment_name: str
|
|
shipment_date: date
|
|
shipment_status: str
|
|
comments: Optional[str]
|
|
contact_person: Optional[ContactPerson]
|
|
return_address: Optional[Address]
|
|
proposal: Optional[Proposal]
|
|
dewars: List[Dewar] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class ShipmentCreate(BaseModel):
|
|
shipment_name: str
|
|
shipment_date: date
|
|
shipment_status: str
|
|
comments: Optional[constr(max_length=200)]
|
|
contact_person_id: int
|
|
return_address_id: int
|
|
proposal_id: int
|
|
dewars: List[DewarCreate] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class UpdateShipmentComments(BaseModel):
|
|
comments: str
|
|
|
|
|
|
class LogisticsEventCreate(BaseModel):
|
|
dewar_qr_code: str
|
|
location_qr_code: str
|
|
transaction_type: str
|
|
|
|
|
|
class SlotSchema(BaseModel):
|
|
id: int
|
|
qr_code: str
|
|
label: str
|
|
qr_base: Optional[str]
|
|
occupied: bool
|
|
needs_refill: bool
|
|
dewar_unique_id: Optional[str]
|
|
dewar_name: Optional[str]
|
|
time_until_refill: Optional[int]
|
|
at_beamline: Optional[bool]
|
|
retrievedTimestamp: Optional[str]
|
|
beamlineLocation: Optional[str]
|
|
shipment_name: Optional[str]
|
|
contact_person: Optional[str]
|
|
local_contact: Optional[str]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class SampleUpdate(BaseModel):
|
|
sample_name: Optional[str] = None
|
|
proteinname: Optional[str] = None
|
|
priority: Optional[int] = None
|
|
position: Optional[int] = None
|
|
comments: Optional[str] = None
|
|
data_collection_parameters: Optional[DataCollectionParameters] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class SetTellPosition(BaseModel):
|
|
puck_name: str
|
|
segment: Optional[str] = Field(
|
|
None,
|
|
pattern="^[A-F]$", # Valid segments are A, B, C, D, E, F
|
|
description="Segment must be one of A, B, C, D, E, or F."
|
|
"Can be null for no tell_position.",
|
|
)
|
|
puck_in_segment: Optional[int] = Field(
|
|
None,
|
|
ge=1,
|
|
le=5,
|
|
description="Puck in segment must be between 1 and 5."
|
|
"Can be null for no tell_position.",
|
|
)
|
|
|
|
@property
|
|
def tell_position(self) -> Optional[str]:
|
|
"""
|
|
Combines `segment` and `puck_in_segment` to generate the `tell_position`.
|
|
If either value is `None`, returns `None` to indicate no `tell_position`.
|
|
"""
|
|
if self.segment and self.puck_in_segment:
|
|
return f"{self.segment}{self.puck_in_segment}"
|
|
return None
|
|
|
|
|
|
class PuckWithTellPosition(BaseModel):
|
|
id: int
|
|
puck_name: str
|
|
puck_type: str
|
|
puck_location_in_dewar: Optional[int]
|
|
dewar_id: Optional[
|
|
int
|
|
] # was changed to optional but probably needs to be not optional
|
|
dewar_name: Optional[
|
|
str
|
|
] # was changed to optional but probably needs to be not optional
|
|
user: str = "e16371"
|
|
samples: Optional[List[Sample]] = None
|
|
tell_position: Optional[str]
|
|
|
|
class Config:
|
|
from_attributes = True
|