from __future__ import annotations
from pathlib import Path
from typing import Optional
from pydantic import BaseModel
from pydantic import Field
from pydantic import field_validator
from ..resources import resource_filename
[docs]
class XrdctProcessorModel(BaseModel, validate_assignment=True):
engine: str = Field(
default="core",
description="Ewoks engine to use for the workflow.",
examples=["core", "dask", "ppf", "orange"],
)
queue: Optional[str] = Field(
default="celery",
description="Ewoks queue to submit the workflow to.",
examples=["celery"],
)
workflow: str = Field(
default="create_nxxrdct.json",
description="Workflow file for creating NX xrdct file.",
examples=["create_nxxrdct.json"],
)
lima_name: Optional[str] = Field(
default=None,
description="Name of the image detector (Ex:pilatus4).",
)
detector_distance: Optional[str] = Field(
default="idx",
description="The variable in the HDF5 file containing the distance between the detector and the sample.",
)
nx_save_path: Optional[str] = Field(
default=None,
description="Path to save Nx file (Optional).",
)
data_portal_upload: bool = Field(
default=False,
description="Boolean to determine if the results should be uploaded to the data portal.",
strict=True,
)
[docs]
@field_validator("engine", mode="before")
def check_engine(cls, v):
if isinstance(v, str):
if v not in ["core", "dask", "ppf", "orange"]:
raise ValueError(
"engine must be either 'core', 'dask', 'ppf', or 'orange'"
)
else:
raise ValueError("engine must be either 'core', 'dask', 'ppf', or 'orange'")
return str(v)
[docs]
@field_validator("queue", mode="before")
def check_queue(cls, v):
if v is not None and not isinstance(v, str):
raise ValueError("queue must be a string or None")
return v
[docs]
@field_validator("workflow", mode="before")
def check_workflow(cls, v):
if not isinstance(v, str):
raise ValueError("workflow must be a string")
if not v.endswith(".json"):
raise ValueError("workflow must end with .json")
if not Path(resource_filename("xrdct", v)).exists():
raise FileNotFoundError(f"workflow file {v} does not exist")
return v
[docs]
@field_validator("lima_name", mode="before")
def check_detector_name(cls, v):
if v is not None and not isinstance(v, str):
raise ValueError("lima_name must be a string or None")
return v
[docs]
@field_validator("detector_distance", mode="before")
def check_detector_distance(cls, v):
if v is not None and not isinstance(v, str):
raise ValueError("detector_distance must be a string or None")
return v
[docs]
@field_validator("nx_save_path", mode="before")
def check_nx_save_path(cls, v):
if v is not None and not isinstance(v, str):
raise ValueError("nx_save_path must be a string or None")
return v