style: ruff format and fix
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / lint (pull_request) Failing after 1m34s
CI / test (3.11) (pull_request) Skipped
CI / test (3.12) (pull_request) Skipped
CI / test (3.13) (pull_request) Skipped
CI / lint (push) Skipped
CI / test (3.11) (push) Skipped
CI / test (3.12) (push) Skipped
CI / test (3.13) (push) Skipped
CI / lint (pull_request) Failing after 1m34s
CI / test (3.11) (pull_request) Skipped
CI / test (3.12) (pull_request) Skipped
CI / test (3.13) (pull_request) Skipped
This commit is contained in:
@@ -40,7 +40,8 @@ the picture if the database wants one.
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
from typing import Any, Optional, Sequence
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
from pydantic import BaseModel, Field
|
||||
@@ -68,7 +69,7 @@ except ImportError: # pragma: no cover
|
||||
z: float = 0.0
|
||||
|
||||
|
||||
__all__ = ["analyse", "GridScanResult", "Centre", "Size", "Counts", "Thresholds"]
|
||||
__all__ = ["Centre", "Counts", "GridScanResult", "Size", "Thresholds", "analyse"]
|
||||
|
||||
#: DAQStatusModel, or anything else carrying the beam: its ``.geom`` is a
|
||||
#: SampleGeometryModel, and a ScanResultPayloadModel has ``beam_size_mm`` itself.
|
||||
@@ -98,10 +99,10 @@ class Centre(BaseModel):
|
||||
nx: float
|
||||
ny: float
|
||||
image_number: int = Field(description="Nearest collected image, for addressing")
|
||||
x_um: Optional[float] = Field(
|
||||
x_um: float | None = Field(
|
||||
None, description="Signed offset from the centre of cell (0,0), along the grid's own axes"
|
||||
)
|
||||
y_um: Optional[float] = None
|
||||
y_um: float | None = None
|
||||
|
||||
def offset_mm(self) -> Coordinate:
|
||||
"""Offset in mm from the centre of cell (0,0), as an aarecommon Coordinate.
|
||||
@@ -123,17 +124,17 @@ class Size(BaseModel):
|
||||
narrower than the beam still reports a size instead of nothing.
|
||||
"""
|
||||
|
||||
x_um: Optional[float] = None
|
||||
y_um: Optional[float] = None
|
||||
z_um: Optional[float] = Field(None, description="From an orthogonal scan, if passed")
|
||||
x_um_deconv: Optional[float] = None
|
||||
y_um_deconv: Optional[float] = None
|
||||
z_um_deconv: Optional[float] = None
|
||||
beam_x_um: Optional[float] = Field(None, description="Beam FWHM used to deconvolve x")
|
||||
beam_y_um: Optional[float] = Field(None, description="Beam FWHM used to deconvolve y")
|
||||
area_um2: Optional[float] = None
|
||||
equiv_diameter_um: Optional[float] = None
|
||||
volume_pl: Optional[float] = Field(None, description="Ellipsoid; needs all three axes")
|
||||
x_um: float | None = None
|
||||
y_um: float | None = None
|
||||
z_um: float | None = Field(None, description="From an orthogonal scan, if passed")
|
||||
x_um_deconv: float | None = None
|
||||
y_um_deconv: float | None = None
|
||||
z_um_deconv: float | None = None
|
||||
beam_x_um: float | None = Field(None, description="Beam FWHM used to deconvolve x")
|
||||
beam_y_um: float | None = Field(None, description="Beam FWHM used to deconvolve y")
|
||||
area_um2: float | None = None
|
||||
equiv_diameter_um: float | None = None
|
||||
volume_pl: float | None = Field(None, description="Ellipsoid; needs all three axes")
|
||||
volume_deconvolved: bool = Field(
|
||||
False, description="True if volume_pl used the beam-removed axes"
|
||||
)
|
||||
@@ -168,50 +169,50 @@ class Counts(BaseModel):
|
||||
n_noise: int = Field(description="Spots present but neither protein nor ice")
|
||||
n_blank: int
|
||||
peak_protein_spots: int
|
||||
mean_protein_in_contour: Optional[float] = None
|
||||
best_resolution_A: Optional[float] = Field(None, description="Best res inside the contour")
|
||||
median_bkg: Optional[float] = None
|
||||
ice_in_contour: Optional[float] = Field(
|
||||
mean_protein_in_contour: float | None = None
|
||||
best_resolution_A: float | None = Field(None, description="Best res inside the contour")
|
||||
median_bkg: float | None = None
|
||||
ice_in_contour: float | None = Field(
|
||||
None, description="Fraction of frames inside the contour showing ice"
|
||||
)
|
||||
n_fragments: int = Field(
|
||||
0, description="Separate blobs at the 50 % level; >1 means more than one region"
|
||||
)
|
||||
ice_fraction: float = Field(description="Fraction of frames showing ice")
|
||||
unit_cell_agreement: Optional[float] = Field(
|
||||
unit_cell_agreement: float | None = Field(
|
||||
None, description="Fraction of indexed frames in the contour sharing one cell"
|
||||
)
|
||||
|
||||
|
||||
class GridScanResult(BaseModel):
|
||||
found: bool
|
||||
reason: Optional[str] = Field(None, description="Why not, when found is False")
|
||||
file_prefix: Optional[str] = None
|
||||
reason: str | None = Field(None, description="Why not, when found is False")
|
||||
file_prefix: str | None = None
|
||||
n_fast: int = 0
|
||||
n_slow: int = 0
|
||||
channel: str = Field("", description="Which spot channel built the map")
|
||||
centre: Optional[Centre] = None
|
||||
size: Optional[Size] = None
|
||||
counts: Optional[Counts] = None
|
||||
unit_cell: Optional[list[float]] = Field(
|
||||
centre: Centre | None = None
|
||||
size: Size | None = None
|
||||
counts: Counts | None = None
|
||||
unit_cell: list[float] | None = Field(
|
||||
None, description="Median cell of the contour, when check_unit_cell is on"
|
||||
)
|
||||
warnings: list[str] = Field(
|
||||
default_factory=list, description="Reasons to distrust a found=True result"
|
||||
)
|
||||
contour_cells: Optional[list[tuple[int, int]]] = Field(
|
||||
contour_cells: list[tuple[int, int]] | None = Field(
|
||||
None,
|
||||
description="(nx, ny) of every collected cell inside the chosen 50 % "
|
||||
"blob -- the decision's own footprint, for drawing it over the sample "
|
||||
"image without recomputing the map",
|
||||
)
|
||||
jpeg: Optional[bytes] = Field(
|
||||
jpeg: bytes | None = Field(
|
||||
None,
|
||||
exclude=True,
|
||||
repr=False,
|
||||
description="The picture, in whichever single style was asked for",
|
||||
)
|
||||
jpeg_cells: Optional[bytes] = Field(
|
||||
jpeg_cells: bytes | None = Field(
|
||||
None,
|
||||
exclude=True,
|
||||
repr=False,
|
||||
@@ -231,10 +232,10 @@ class Thresholds(BaseModel):
|
||||
"the bkg-only gate then rejects the true crystal",
|
||||
)
|
||||
ice_spots: int = Field(5, description="spots_ice at or above this counts as ice")
|
||||
ice_ring: Optional[float] = Field(
|
||||
ice_ring: float | None = Field(
|
||||
None, description="Also call ice if scan_result 'ice' exceeds this"
|
||||
)
|
||||
beam_um: Optional[float] = Field(
|
||||
beam_um: float | None = Field(
|
||||
None,
|
||||
description="Beam FWHM in um, used only when no DAQ status is passed. "
|
||||
"A single number is taken as a square beam",
|
||||
@@ -290,7 +291,7 @@ def _upsample(a: np.ndarray, f: int) -> np.ndarray:
|
||||
return np.asarray(im.resize((nx * fx, ny * fy), Image.BICUBIC), dtype=float)
|
||||
|
||||
|
||||
def _norm(a: np.ndarray, floor: float = 0.0, ref: Optional[np.ndarray] = None) -> np.ndarray:
|
||||
def _norm(a: np.ndarray, floor: float = 0.0, ref: np.ndarray | None = None) -> np.ndarray:
|
||||
"""Normalise for colour against the channel's own range, but never below floor.
|
||||
|
||||
``ref`` supplies the range: the bounds come from the frames actually measured,
|
||||
@@ -378,7 +379,7 @@ _BEAM_PATHS = (
|
||||
)
|
||||
|
||||
|
||||
def _beam_um(obj: Optional[BeamSource]) -> tuple[Optional[float], Optional[float]]:
|
||||
def _beam_um(obj: BeamSource | None) -> tuple[float | None, float | None]:
|
||||
"""Beam FWHM (x, y) in micrometres, dug out of a DAQ status or scan payload.
|
||||
|
||||
The beam is a Coordinate, not one number: on a beamline with an 80 x 20 um
|
||||
@@ -412,7 +413,7 @@ def _beam_um(obj: Optional[BeamSource]) -> tuple[Optional[float], Optional[float
|
||||
return None, None
|
||||
|
||||
|
||||
def _deconvolve(fwhm: Optional[float], beam: Optional[float]) -> Optional[float]:
|
||||
def _deconvolve(fwhm: float | None, beam: float | None) -> float | None:
|
||||
"""Quadrature removal of the beam. None when the beam swallows the feature."""
|
||||
if fwhm is None or beam is None:
|
||||
return fwhm
|
||||
@@ -495,11 +496,11 @@ def _classify(F: dict, th: Thresholds) -> Counts:
|
||||
|
||||
def analyse(
|
||||
scan_result: ScanResult,
|
||||
grid_scan: Optional[GridScan] = None,
|
||||
grid_scan: GridScan | None = None,
|
||||
*,
|
||||
daq: Optional[BeamSource] = None,
|
||||
thresholds: Optional[Thresholds] = None,
|
||||
z_um: Optional[float] = None,
|
||||
daq: BeamSource | None = None,
|
||||
thresholds: Thresholds | None = None,
|
||||
z_um: float | None = None,
|
||||
jpeg: bool = True,
|
||||
) -> GridScanResult:
|
||||
"""Analyse one grid scan.
|
||||
@@ -649,7 +650,7 @@ def analyse(
|
||||
w_cells = ((xx.max() - xx.min() + 1) / mx * nx_n, (yy.max() - yy.min() + 1) / my * ny_n)
|
||||
|
||||
# the nearest image actually collected, for the DAQ to address
|
||||
iy, ix = int(round(np.clip(gy, 0, ny_n - 1))), int(round(np.clip(gx, 0, nx_n - 1)))
|
||||
iy, ix = round(np.clip(gy, 0, ny_n - 1)), round(np.clip(gx, 0, nx_n - 1))
|
||||
num = int(F["number"][iy, ix])
|
||||
if num < 0: # that cell was never collected
|
||||
yy2, xx2 = np.nonzero(F["number"] >= 0)
|
||||
@@ -751,7 +752,7 @@ def _edge(mask: np.ndarray, width: int = 1) -> np.ndarray:
|
||||
return mask & ~core
|
||||
|
||||
|
||||
def _pictures(F, inside, centre, res: "GridScanResult", th: Thresholds, obj_level: float) -> None:
|
||||
def _pictures(F, inside, centre, res: GridScanResult, th: Thresholds, obj_level: float) -> None:
|
||||
"""Fill res.jpeg, and res.jpeg_cells too when both styles were asked for."""
|
||||
want = ("smooth", "cells") if th.jpeg_style == "both" else (th.jpeg_style,)
|
||||
res.jpeg = _render(F, inside, centre, res, th, obj_level, want[0])
|
||||
@@ -760,7 +761,7 @@ def _pictures(F, inside, centre, res: "GridScanResult", th: Thresholds, obj_leve
|
||||
|
||||
|
||||
def _render(
|
||||
F, inside, centre, res: "GridScanResult", th: Thresholds, obj_level: float, style: str
|
||||
F, inside, centre, res: GridScanResult, th: Thresholds, obj_level: float, style: str
|
||||
) -> bytes:
|
||||
"""Map, contours, crosshair and a one-line caption, as JPEG.
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ The pictures are excluded from the result's serialization by the model itself;
|
||||
the DAQ ships them through the ordinary image pipeline.
|
||||
"""
|
||||
|
||||
from typing import Any, Optional
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
@@ -25,10 +25,10 @@ class GridScanDecision(BaseModel):
|
||||
description="Which analysis took the decision, e.g. a "
|
||||
"GridscanAnalysisMode value such as 'jfjoch_gridscan_union'"
|
||||
)
|
||||
algorithm_version: Optional[str] = Field(
|
||||
algorithm_version: str | None = Field(
|
||||
None, description="aarecommon version (or commit) the DAQ ran"
|
||||
)
|
||||
thresholds: Optional[dict[str, Any]] = Field(
|
||||
thresholds: dict[str, Any] | None = Field(
|
||||
None,
|
||||
description="Thresholds.model_dump() used at decision time, so a "
|
||||
"stored decision states its own tuning",
|
||||
|
||||
@@ -36,9 +36,7 @@ def test_decision_round_trips_without_pictures():
|
||||
r = analyse(_synthetic_scan(), {"step_x_um": 10.0, "step_y_um": 10.0}, thresholds=th)
|
||||
assert r.found
|
||||
assert r.contour_cells, "the chosen 50 % blob must ship its cells"
|
||||
assert all(
|
||||
0 <= x < r.n_fast and 0 <= y < r.n_slow for x, y in r.contour_cells
|
||||
)
|
||||
assert all(0 <= x < r.n_fast and 0 <= y < r.n_slow for x, y in r.contour_cells)
|
||||
# the centre lies within the contour's bounding box
|
||||
xs = [x for x, _ in r.contour_cells]
|
||||
ys = [y for _, y in r.contour_cells]
|
||||
|
||||
Reference in New Issue
Block a user