from __future__ import annotations
from pathlib import Path
from typing import List
from typing import Literal
from typing import Optional
from typing import Tuple
from typing import Union
from pydantic import BaseModel
from pydantic import Field
from pydantic import field_validator
from ..resources import resource_filename
[docs]
class TomoProcessorModel(BaseModel, validate_assignment=True):
slice_reconstruction_workflow: str = Field(
default="slice_reconstruction.json",
description="Workflow file for slice reconstruction",
examples=["slice_reconstruction.json"],
)
volume_reconstruction_workflow: str = Field(
default="volume_reconstruction.json",
description="Workflow file for volume reconstruction",
examples=["volume_reconstruction.json"],
)
queue: Optional[str] = Field(
default=None,
description="Ewoks queue to submit the workflow to",
examples=["tomo_queue"],
)
nabu_config_file: str = Field(default=resource_filename("tomo", "nabu.conf"))
slice_index: Union[Literal["first", "middle", "last"], int] = "middle"
phase_retrieval_method: Union[Literal["CTF", "Paganin", "None"], None] = "None"
delta_beta: Union[str, float, int] = "100"
offset_mm: Union[str, float, int] = "0"
cor_algorithm: Union[
int,
float,
Literal[
"centered",
"sliding-window",
"growing-window",
"composite-coarse-to-fine",
"octave-accurate",
],
] = "sliding-window"
show_last_slice: bool = Field(
default=False,
description="Display last reconstructed slice in Flint",
)
volume_reconstruction: bool = False
output_format: Literal["tiff", "hdf5", "jp2", "edf", "vol"] = "hdf5"
slice_index_range: Union[Literal["all"], Tuple[int, int], List[int]] = "all"
[docs]
@field_validator("slice_reconstruction_workflow")
def check_slice_workflow(cls, v):
if not isinstance(v, str):
raise ValueError("slice_reconstruction_workflow must be a string")
if not v.endswith(".json"):
raise ValueError("slice_reconstruction_workflow must end with .json")
if not Path(resource_filename("tomo", v)).exists():
raise FileNotFoundError(f"Workflow file {v} does not exist")
return v
[docs]
@field_validator("volume_reconstruction_workflow")
def check_volume_workflow(cls, v):
if not isinstance(v, str):
raise ValueError("volume_reconstruction_workflow must be a string")
if not v.endswith(".json"):
raise ValueError("volume_reconstruction_workflow must end with .json")
if not Path(resource_filename("tomo", v)).exists():
raise FileNotFoundError(f"Workflow file {v} does not exist")
return v
[docs]
@field_validator("queue")
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("nabu_config_file")
def check_nabu_config_file(cls, v):
if v is not None and not isinstance(v, str):
raise ValueError("nabu_config_file must be a string or None")
if v and not Path(v).exists():
raise FileNotFoundError(f"Nabu config file {v} does not exist")
return v
[docs]
@field_validator("slice_index")
def check_slice_index(cls, v):
if isinstance(v, str):
if v not in ["first", "last", "middle"]:
raise ValueError(
"slice_index must be 'first', 'last', 'middle' or an integer"
)
elif not isinstance(v, int):
raise ValueError(
"slice_index must be an integer or one of 'first', 'last', 'middle'"
)
return str(v)
[docs]
@field_validator("cor_algorithm")
def check_cor_algorithm(cls, v):
valid_positions = [
"centered",
"sliding-window",
"global",
"growing-window",
"sino-coarse-to-fine",
"composite-coarse-to-fine",
"octave-accurate",
]
if isinstance(v, str):
if v not in valid_positions:
raise ValueError(f"cor_algorithm must be one of {valid_positions}")
return v
[docs]
@field_validator("phase_retrieval_method")
def check_phase_method(cls, v):
if v is None:
return "None"
elif v not in ["CTF", "Paganin", "None"]:
raise ValueError("phase_retrieval_method must be 'CTF', 'Paganin' or None")
return v
[docs]
@field_validator("delta_beta")
def check_positive_delta_beta(cls, v):
if float(v) <= 0:
raise ValueError("delta_beta must be positive")
return str(v)
[docs]
@field_validator("offset_mm")
def validate_offset(cls, v):
if v is None:
return 0.0
try:
return float(v)
except (TypeError, ValueError):
raise ValueError("offset must be numeric")
[docs]
@field_validator("slice_index_range")
def check_slice_index_range(cls, v):
if v != "all":
if (
not isinstance(v, tuple)
or not isinstance(v, list)
or len(v) != 2
or not all(isinstance(i, int) for i in v)
or v[0] < 0
or v[1] < v[0]
):
raise ValueError(
"slice_index_range must be 'all' or a tuple of two non-negative integers (z_bottom, z_top) with z_top >= z_bottom"
)
return v