diff --git a/lambda/src/data_hub_lambda/cli.py b/lambda/src/data_hub_lambda/cli.py index 2177a2ea..3b8cf920 100644 --- a/lambda/src/data_hub_lambda/cli.py +++ b/lambda/src/data_hub_lambda/cli.py @@ -104,17 +104,49 @@ def hina(file: Path, output_dir: Path | None) -> None: default=None, help="Directory for the exported JPG (default: same directory as FILE).", ) -def epson_scanner(file: Path, output_dir: Path | None) -> None: +@click.option( + "--detect-colonies", + is_flag=True, + default=False, + help="Run colony detection and phenotyping on each detected plate.", +) +def epson_scanner(file: Path, output_dir: Path | None, detect_colonies: bool) -> None: """Process an Epson V700 Scanner TIFF file. - Resizes the high-resolution scan to a web-friendly JPEG preview and - extracts TIFF metadata. + Detects agar plates inside gold frames, draws bounding-box overlays, + resizes to a web-friendly JPEG preview, and extracts TIFF metadata. """ - from data_hub_lambda.epson_v700_scanner.image_processing import TIFFToJPEGConverter + from data_hub_lambda.epson_v700_scanner.image_processing import TiffProcessor - converter = TIFFToJPEGConverter(file) - converter.load() - jpg_path = converter.export_jpg() + processor = TiffProcessor(file) + processor.load() + + pipeline = None + if detect_colonies: + from data_hub_lambda.epson_v700_scanner.colony_detection import ( + export_colony_csv, + run_colony_pipeline, + ) + + processor.detect_plates() + plate_crops = processor.crop_plates() + if not plate_crops: + click.echo("No plates detected — skipping colony detection.") + else: + pipeline = run_colony_pipeline(plate_crops, dpi=processor.dpi) + for i, result in enumerate(pipeline.results): + click.echo(f"\nPlate {i + 1}:") + click.echo(json.dumps(result.summary(), indent=2)) + + dest_dir = output_dir or file.parent + dest_dir.mkdir(parents=True, exist_ok=True) + csv_path = dest_dir / f"{file.stem}_colonies.csv" + export_colony_csv(pipeline.to_dataframes(), csv_path) + click.echo(f"\nColony CSV: {csv_path}") + + jpg_path = processor.export_jpg( + colony_results=pipeline.results if pipeline else None, + ) if output_dir is not None: output_dir.mkdir(parents=True, exist_ok=True) @@ -124,7 +156,7 @@ def epson_scanner(file: Path, output_dir: Path | None) -> None: click.echo(f"Exported JPG: {jpg_path}") - metadata = converter.parse_metadata() + metadata = processor.parse_metadata() click.echo(json.dumps(metadata, indent=2)) diff --git a/lambda/src/data_hub_lambda/epson_v700_scanner/colony_detection.py b/lambda/src/data_hub_lambda/epson_v700_scanner/colony_detection.py new file mode 100644 index 00000000..9b3632bf --- /dev/null +++ b/lambda/src/data_hub_lambda/epson_v700_scanner/colony_detection.py @@ -0,0 +1,399 @@ +"""Colony detection and phenotyping from cropped agar-plate images. + +Operates on individual plate crops produced by +:class:`~data_hub_lambda.epson_v700_scanner.image_processing.TiffProcessor`. + +Pipeline +-------- +1. Crop a fixed pixel margin to remove plate edges / frame artefacts. +2. Optimise contrast (Euclidean distance from estimated background colour). +3. Decide whether colonies are present (contrast above noise floor). +4. Difference-of-Gaussians band-pass filter with 10th-percentile subtraction. +5. Otsu threshold to produce a binary colony mask. +6. Morphological cleanup: remove sub-threshold objects and border-touching + regions, then label and measure surviving colonies. +""" + +from __future__ import annotations +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd +import skimage as ski +from numpy.typing import NDArray + +logger = logging.getLogger(__name__) + +MARGIN_PX = 200 +"""Fixed pixel margin cropped from each edge before colony detection.""" + +_DOG_LOW_SIGMA = 1.0 +"""Low sigma for the Difference-of-Gaussians band-pass filter.""" + +_DOG_HIGH_SIGMA = 128.0 +"""High sigma for the Difference-of-Gaussians band-pass filter.""" + +_PERCENTILE_FLOOR = 10.0 +"""Percentile used for baseline subtraction after DoG filtering.""" + +_CONTRAST_PRESENCE_THRESHOLD = 20.0 +"""Minimum 99.5th-percentile contrast value to declare colonies present.""" + +_MIN_COLONY_AREA_MM2 = 0.05 +"""Colonies smaller than this (in mm²) are discarded as noise.""" + + +@dataclass +class ColonyProperties: + """Measured properties for a single colony (physical units, plate-relative).""" + + label: int + area_mm2: float + centroid_row_mm: float + centroid_col_mm: float + bbox_mm: tuple[float, float, float, float] + eccentricity: float + equivalent_diameter_mm: float + mean_rgb: tuple[float, float, float] + + +_DATAFRAME_COLUMNS = [ + "plate_index", + "label", + "area_mm2", + "centroid_row_mm", + "centroid_col_mm", + "bbox_min_row_mm", + "bbox_min_col_mm", + "bbox_max_row_mm", + "bbox_max_col_mm", + "eccentricity", + "equivalent_diameter_mm", + "mean_r", + "mean_g", + "mean_b", +] + + +@dataclass +class ColonyDetectionResult: + """Full result of the colony-detection pipeline for a single plate.""" + + cropped: NDArray[Any] + contrast: NDArray[np.floating[Any]] + has_colonies: bool + mask: NDArray[np.bool_] + colonies: list[ColonyProperties] = field(default_factory=list) + + def summary(self) -> dict[str, Any]: + """Return a JSON-serialisable summary dict.""" + return { + "has_colonies": self.has_colonies, + "colony_count": len(self.colonies), + "colonies": [ + { + "label": c.label, + "area_mm2": round(c.area_mm2, 4), + "centroid_mm": [ + round(c.centroid_row_mm, 3), + round(c.centroid_col_mm, 3), + ], + "bbox_mm": [round(v, 3) for v in c.bbox_mm], + "eccentricity": round(c.eccentricity, 4), + "equivalent_diameter_mm": round(c.equivalent_diameter_mm, 4), + "mean_rgb": [round(v, 1) for v in c.mean_rgb], + } + for c in self.colonies + ], + } + + def to_dataframe(self, plate_index: int = 0) -> pd.DataFrame: + """Return a :class:`~pandas.DataFrame` with one row per colony.""" + if not self.colonies: + return pd.DataFrame(columns=_DATAFRAME_COLUMNS) + rows = [] + for c in self.colonies: + rows.append( + { + "plate_index": plate_index, + "label": c.label, + "area_mm2": round(c.area_mm2, 4), + "centroid_row_mm": round(c.centroid_row_mm, 3), + "centroid_col_mm": round(c.centroid_col_mm, 3), + "bbox_min_row_mm": round(c.bbox_mm[0], 3), + "bbox_min_col_mm": round(c.bbox_mm[1], 3), + "bbox_max_row_mm": round(c.bbox_mm[2], 3), + "bbox_max_col_mm": round(c.bbox_mm[3], 3), + "eccentricity": round(c.eccentricity, 4), + "equivalent_diameter_mm": round(c.equivalent_diameter_mm, 4), + "mean_r": round(c.mean_rgb[0], 1), + "mean_g": round(c.mean_rgb[1], 1), + "mean_b": round(c.mean_rgb[2], 1), + } + ) + return pd.DataFrame(rows) + + +@dataclass +class ColonyPipelineResult: + """Aggregated colony-detection results across all plates in a scan.""" + + results: list[ColonyDetectionResult] + + @property + def summaries(self) -> list[dict[str, Any]]: + return [r.summary() for r in self.results] + + def to_dataframes(self) -> list[pd.DataFrame]: + return [r.to_dataframe(plate_index=i + 1) for i, r in enumerate(self.results)] + + +# ------------------------------------------------------------------ +# Pipeline steps +# ------------------------------------------------------------------ + + +def crop_margin(image: NDArray[Any], margin_px: int = MARGIN_PX) -> NDArray[Any]: + """Remove a fixed pixel margin from each edge of *image*. + + Args: + image: (H, W) or (H, W, C) array. + margin_px: Number of pixels to remove from each side. + + Returns: + Cropped view of the original array. + """ + h, w = image.shape[:2] + return image[margin_px : h - margin_px, margin_px : w - margin_px] + + +def optimize_colony_contrast(image: NDArray[Any]) -> NDArray[np.floating[Any]]: + """Convert to a single channel that maximises colony-background contrast. + + Per-pixel Euclidean distance from the estimated background colour + (median of non-zero pixels). Colonies of any colour appear bright + against a near-zero background. + + Args: + image: (H, W) or (H, W, C) array. + + Returns: + (H, W) float64 distance-from-background image. + """ + if image.ndim == 2: + return image.astype(np.float64) + + pixels = image.reshape(-1, image.shape[-1]) + nonzero = np.any(pixels > 0, axis=1) + background: NDArray[np.floating[Any]] + if nonzero.any(): + background = np.median(pixels[nonzero], axis=0) + else: + background = np.median(pixels, axis=0) + + diff = image.astype(np.float64) - background + distance: NDArray[np.floating[Any]] = np.sqrt(np.sum(diff**2, axis=-1)) + distance[~np.any(image > 0, axis=-1)] = 0.0 + return distance + + +def detect_colony_presence( + contrast: NDArray[np.floating[Any]], + threshold: float = _CONTRAST_PRESENCE_THRESHOLD, +) -> bool: + """Return ``True`` if there is enough contrast to indicate colonies. + + Uses the 99.5th percentile of the contrast image; plates without + colonies have near-uniform background. + """ + if contrast.size == 0: + return False + p = float(np.percentile(contrast, 99.5)) + logger.debug("Colony-presence p99.5 contrast = %.2f (threshold %.2f)", p, threshold) + return p > threshold + + +def smooth( + contrast: NDArray[np.floating[Any]], + low_sigma: float = _DOG_LOW_SIGMA, + high_sigma: float = _DOG_HIGH_SIGMA, + percentile: float = _PERCENTILE_FLOOR, +) -> NDArray[np.floating[Any]]: + """Apply Difference-of-Gaussians band-pass filter with percentile subtraction. + + The DoG filter retains structures between *low_sigma* and *high_sigma* + scale, suppressing both high-frequency noise and low-frequency background + gradients. A 10th-percentile baseline is then subtracted and negative + values are clipped to zero so the result stays non-negative for Otsu + thresholding. + """ + dog: NDArray[np.floating[Any]] = ski.filters.difference_of_gaussians( + contrast, + low_sigma=low_sigma, + high_sigma=high_sigma, + ) + floor = np.percentile(dog, percentile) + result: NDArray[np.floating[Any]] = np.clip(dog - floor, 0, None) + return result + + +def threshold_colonies(smoothed: NDArray[np.floating[Any]]) -> NDArray[np.bool_]: + """Otsu threshold to produce a binary colony mask.""" + thresh = ski.filters.threshold_otsu(smoothed) + mask: NDArray[np.bool_] = smoothed > thresh + return mask + + +def measure_colonies( + mask: NDArray[np.bool_], + rgb_image: NDArray[np.uint8], + dpi: int, + margin_px: int = MARGIN_PX, +) -> tuple[list[ColonyProperties], NDArray[np.bool_]]: + """Label connected components and extract per-colony measurements. + + All spatial measurements are returned in millimetres. Centroids and + bounding boxes are expressed relative to the plate crop origin (i.e. + the margin offset is added back before conversion). + + Args: + mask: Binary colony mask (margin-cropped coordinate space). + rgb_image: RGB uint8 image with the same shape as *mask* + (margin-cropped). Used as ``intensity_image`` so that + ``regionprops`` can compute per-colony mean colour. + dpi: Image resolution in dots-per-inch. + margin_px: Pixel margin that was removed from the plate crop. + + Returns: + A tuple of (colony list, cleaned mask). The cleaned mask has + border-touching regions removed. + """ + mm_per_px = 25.4 / dpi + min_area_px = int(np.ceil(_MIN_COLONY_AREA_MM2 / mm_per_px**2)) + + mask = ski.morphology.remove_small_objects(mask, max_size=min_area_px) + cleared_mask = ski.segmentation.clear_border(mask) + labels = ski.measure.label(cleared_mask) + regions = ski.measure.regionprops(labels, intensity_image=rgb_image) + + colonies: list[ColonyProperties] = [] + for region in regions: + row_px = float(region.centroid[0]) + margin_px + col_px = float(region.centroid[1]) + margin_px + + min_r, min_c, max_r, max_c = region.bbox + bbox_mm = ( + (min_r + margin_px) * mm_per_px, + (min_c + margin_px) * mm_per_px, + (max_r + margin_px) * mm_per_px, + (max_c + margin_px) * mm_per_px, + ) + + rgb_mean = region.intensity_mean + mean_rgb = (float(rgb_mean[0]), float(rgb_mean[1]), float(rgb_mean[2])) + + colonies.append( + ColonyProperties( + label=int(region.label), + area_mm2=float(region.area) * mm_per_px**2, + centroid_row_mm=row_px * mm_per_px, + centroid_col_mm=col_px * mm_per_px, + bbox_mm=bbox_mm, + eccentricity=float(region.eccentricity), + equivalent_diameter_mm=float(region.equivalent_diameter_area) * mm_per_px, + mean_rgb=mean_rgb, + ) + ) + + colonies.sort(key=lambda c: c.area_mm2, reverse=True) + return colonies, cleared_mask + + +# ------------------------------------------------------------------ +# Visualisation & export +# ------------------------------------------------------------------ + + +def export_colony_csv( + frames: list[pd.DataFrame], + path: Path, +) -> Path: + """Concatenate per-plate DataFrames and write a CSV.""" + if frames: + combined = pd.concat(frames, ignore_index=True) + else: + combined = pd.DataFrame(columns=_DATAFRAME_COLUMNS) + combined.to_csv(path, index=False) + logger.debug("Wrote colony CSV: %s", path) + return path + + +# ------------------------------------------------------------------ +# Orchestrator +# ------------------------------------------------------------------ + + +def detect_colonies(plate_image: NDArray[Any], dpi: int) -> ColonyDetectionResult: + """Run the full colony-detection pipeline on a single plate crop. + + Args: + plate_image: RGB uint8 plate image produced by + :class:`~data_hub_lambda.epson_v700_scanner.image_processing.TiffProcessor`. + dpi: Image resolution in dots-per-inch, used to convert pixel + measurements to millimetres. + + Returns: + A :class:`ColonyDetectionResult` with all intermediate arrays + and measured colony properties. + """ + cropped = crop_margin(plate_image) + contrast = optimize_colony_contrast(cropped) + has_colonies = detect_colony_presence(contrast) + + if not has_colonies: + logger.info("No colonies detected (low contrast).") + return ColonyDetectionResult( + cropped=cropped, + contrast=contrast, + has_colonies=False, + mask=np.zeros(contrast.shape, dtype=bool), + ) + + smoothed = smooth(contrast) + raw_mask = threshold_colonies(smoothed) + colonies, cleaned_mask = measure_colonies(raw_mask, rgb_image=cropped, dpi=dpi) + + logger.info("Detected %d colony/ies.", len(colonies)) + return ColonyDetectionResult( + cropped=cropped, + contrast=contrast, + has_colonies=True, + mask=cleaned_mask, + colonies=colonies, + ) + + +def run_colony_pipeline( + plate_crops: list[NDArray[np.uint8]], + dpi: int, +) -> ColonyPipelineResult: + """Run colony detection on every plate crop and return aggregated results. + + This is the single entry point for colony detection across all plates in + a scan. Both the Lambda handler and the CLI delegate to this function. + + Args: + plate_crops: RGB uint8 plate images from + :meth:`~data_hub_lambda.epson_v700_scanner.image_processing.TiffProcessor.crop_plates`. + dpi: Image resolution in dots-per-inch. + """ + results = [detect_colonies(crop, dpi=dpi) for crop in plate_crops] + logger.info( + "Colony pipeline complete for %d plate(s): %s", + len(results), + [r.summary()["colony_count"] for r in results], + ) + return ColonyPipelineResult(results=results) diff --git a/lambda/src/data_hub_lambda/epson_v700_scanner/image_processing.py b/lambda/src/data_hub_lambda/epson_v700_scanner/image_processing.py index bffa601c..c3fdd04a 100644 --- a/lambda/src/data_hub_lambda/epson_v700_scanner/image_processing.py +++ b/lambda/src/data_hub_lambda/epson_v700_scanner/image_processing.py @@ -3,19 +3,45 @@ from pathlib import Path from typing import Any +import imageio.v3 as iio import numpy as np import skimage as ski import tifffile from numpy.typing import NDArray +from data_hub_lambda.epson_v700_scanner.colony_detection import ( + ColonyDetectionResult, +) + logger = logging.getLogger(__name__) JPEG_QUALITY = 85 -MAX_DIMENSION = 1000 +MAX_DIMENSION = 2000 _TIFF_SUFFIXES = {".tif", ".tiff"} +_PlateBox = tuple[int, int, int, int] + +_GOLD_HUE_LOW = 0.06 +_GOLD_HUE_HIGH = 0.18 +_GOLD_SAT_MIN = 0.25 +_GOLD_VAL_MIN = 0.35 + +_MIN_AREA_FRACTION = 0.05 +_MIN_EXTENT = 0.85 + +_DETECTION_DOWNSAMPLE = 4 +_CLOSING_RADIUS = 5 + +_OVERLAY_COLOR: tuple[int, int, int] = (0, 255, 0) +_OVERLAY_THICKNESS = 8 + +_COLONY_BBOX_COLOR: tuple[int, int, int] = (255, 255, 255) +_COLONY_BBOX_THICKNESS = 3 +_COLONY_LABEL_FONT_SIZE = 28 +_COLONY_LABEL_MAX_COUNT = 300 + # PhotometricInterpretation values: 2 = RGB, others (0, 1, 3) are grayscale or # palette and are treated as B&W for our display purposes. _PHOTOMETRIC_RGB = 2 @@ -30,7 +56,7 @@ def _derive_dpi(x_resolution: Any) -> int | None: return None if denominator == 0: return None - return int(numerator / denominator) + return round(numerator / denominator) def _derive_color_mode(samples_per_pixel: Any, photometric_interpretation: Any) -> str | None: @@ -63,12 +89,26 @@ def _derive_color_mode(samples_per_pixel: Any, photometric_interpretation: Any) } -class TIFFToJPEGConverter: - """Converts high-resolution TIFF scans to resized JPEG images.""" +class TiffProcessor: + """Processes high-resolution TIFF scans from the Epson V700 flatbed scanner.""" def __init__(self, path: Path) -> None: self.path = path self._intensities: NDArray[Any] | None = None + self._dpi: int | None = None + self._rgb_cache: NDArray[np.uint8] | None = None + self.plate_boxes: list[_PlateBox] | None = None + + @property + def dpi(self) -> int: + """Integer DPI derived from the TIFF ``XResolution`` tag.""" + if self._dpi is None: + with tifffile.TiffFile(self.path) as tif: + x_res_tag = tif.pages.first.tags.get("XResolution") + self._dpi = _derive_dpi(x_res_tag.value if x_res_tag else None) + if self._dpi is None: + raise ValueError(f"Cannot determine DPI for {self.path}") + return self._dpi def load(self) -> None: if not self.path.exists(): @@ -84,25 +124,51 @@ def intensities(self) -> NDArray[Any]: raise RuntimeError("Call load() first.") return self._intensities - def export_jpg(self) -> Path: - """Resize the loaded TIFF and write a JPEG next to the source file.""" - img = self._to_rgb_uint8(self.intensities) + def export_jpg( + self, + colony_results: list[ColonyDetectionResult] | None = None, + ) -> Path: + """Detect plates, draw overlays, resize, and write a JPEG. + + If no gold frames are detected the full image is exported as-is + (the pre-detection fallback behaviour). + + Args: + colony_results: Optional per-plate detection results (one per + entry in ``plate_boxes``). When provided, colony bounding + boxes and labels are drawn on the export image. + """ + img = self._get_rgb_uint8().copy() + if self.plate_boxes is None: + self.detect_plates(img) + + if self.plate_boxes: + img = self._draw_plate_overlays(img, self.plate_boxes) + if colony_results: + img = self._draw_colony_bboxes( + img, + self.plate_boxes, + colony_results, + dpi=self.dpi, + ) + img = self._resize(img) jpg_path = self.path.parent / f"{self.path.stem}.jpg" - ski.io.imsave(str(jpg_path), img, quality=JPEG_QUALITY) + iio.imwrite(jpg_path, img, quality=JPEG_QUALITY) return jpg_path def parse_metadata(self) -> dict[str, Any]: """Extract TIFF tags as a flat string-keyed dict. - In addition to the raw TIFF tags, this also emits two derived - scalar fields used by the web UI for filtering and display: + Derived fields: - - ``dpi``: integer DPI computed from ``XResolution`` (a (numerator, - denominator) rational). For Epson V700 scans this is 300 or 600. - - ``color_mode``: ``"rgb"`` or ``"bw"``, inferred from - ``SamplesPerPixel`` (preferred) or ``PhotometricInterpretation``. + - ``dpi``: integer DPI from ``XResolution``. + - ``color_mode``: ``"rgb"`` or ``"bw"``. + - ``plate_count``: number of detected plates (runs detection if + not already performed). + - ``plate_boxes``: list of ``[min_row, min_col, max_row, max_col]`` + bounding boxes in original-image coordinates. """ metadata: dict[str, Any] = {} with tifffile.TiffFile(self.path) as tif: @@ -129,23 +195,215 @@ def parse_metadata(self) -> dict[str, Any]: if color_mode is not None: metadata["color_mode"] = color_mode + if self.plate_boxes is None: + self.detect_plates() + assert self.plate_boxes is not None # populated by detect_plates + metadata["plate_count"] = len(self.plate_boxes) + metadata["plate_boxes"] = [list(b) for b in self.plate_boxes] + return metadata + # ------------------------------------------------------------------ + # Plate detection + # ------------------------------------------------------------------ + + def detect_plates(self, img: NDArray[np.uint8] | None = None) -> None: + """Detect agar plates inside gold 3D-printed frames. + + Runs detection on a downsampled copy for speed, then scales + bounding boxes back to original coordinates. Results are stored + in ``self.plate_boxes`` as ``(min_row, min_col, max_row, max_col)`` + tuples sorted left-to-right. + + Args: + img: Optional RGB uint8 image. When *None* the image is + derived from ``self.intensities``. + """ + if img is None: + img = self._get_rgb_uint8() + h_orig, w_orig = img.shape[:2] + s = _DETECTION_DOWNSAMPLE + small: NDArray[np.uint8] = img[::s, ::s] + + hsv = ski.color.rgb2hsv(small) + + gold_mask: NDArray[np.bool_] = ( + (hsv[:, :, 0] >= _GOLD_HUE_LOW) + & (hsv[:, :, 0] <= _GOLD_HUE_HIGH) + & (hsv[:, :, 1] >= _GOLD_SAT_MIN) + & (hsv[:, :, 2] >= _GOLD_VAL_MIN) + ) + + selem = ski.morphology.disk(_CLOSING_RADIUS) + gold_mask = ski.morphology.closing(gold_mask, selem) + + inverted = ~gold_mask + labels = ski.measure.label(inverted) + regions = ski.measure.regionprops(labels) + + h_small, w_small = small.shape[:2] + min_area = h_small * w_small * _MIN_AREA_FRACTION + + boxes: list[_PlateBox] = [] + for region in regions: + if region.area < min_area: + continue + if region.extent < _MIN_EXTENT: + continue + + min_row, min_col, max_row, max_col = region.bbox + touches_border = ( + min_row == 0 or min_col == 0 or max_row == h_small or max_col == w_small + ) + if touches_border: + continue + + boxes.append( + ( + min(min_row * s, h_orig), + min(min_col * s, w_orig), + min(max_row * s, h_orig), + min(max_col * s, w_orig), + ) + ) + + boxes.sort(key=lambda b: b[1]) + self.plate_boxes = boxes + logger.debug("Detected %d plate(s): %s", len(boxes), boxes) + + @staticmethod + def _draw_plate_overlays( + img: NDArray[np.uint8], + boxes: list[_PlateBox], + ) -> NDArray[np.uint8]: + """Draw coloured rectangle outlines on a copy of the image. + + Assumes non-overlapping boxes; overlapping boxes would cause + later interior restores to overwrite earlier overlay lines. + """ + out = img.copy() + h, w = out.shape[:2] + t = _OVERLAY_THICKNESS + for min_row, min_col, max_row, max_col in boxes: + r0 = max(min_row - t, 0) + c0 = max(min_col - t, 0) + r1 = min(max_row + t, h) + c1 = min(max_col + t, w) + out[r0:r1, c0:c1] = _OVERLAY_COLOR + out[min_row:max_row, min_col:max_col] = img[min_row:max_row, min_col:max_col] + return out + + @staticmethod + def _draw_colony_bboxes( + img: NDArray[np.uint8], + boxes: list[_PlateBox], + colony_results: list[ColonyDetectionResult], + dpi: int, + ) -> NDArray[np.uint8]: + """Draw colony bounding boxes and labels onto *img* for each plate. + + Uses the pre-computed :class:`ColonyProperties` from each + detection result rather than re-labelling the masks. Bounding + boxes (stored in mm, plate-crop-relative) are converted back to + full-image pixel coordinates. + + Uses matplotlib for anti-aliased rectangle and text rendering. + """ + from matplotlib.backends.backend_agg import FigureCanvasAgg + from matplotlib.figure import Figure + from matplotlib.patches import Rectangle + + mm_per_px = 25.4 / dpi + img_h, img_w = img.shape[:2] + fig_dpi = 100 + fig = Figure(figsize=(img_w / fig_dpi, img_h / fig_dpi), dpi=fig_dpi) + canvas = FigureCanvasAgg(fig) + ax = fig.add_subplot(1, 1, 1) + ax.imshow(img) + ax.set_axis_off() + ax.set_xlim(0, img_w) + ax.set_ylim(img_h, 0) + + bbox_color = tuple(c / 255.0 for c in _COLONY_BBOX_COLOR) + + for box, result in zip(boxes, colony_results, strict=True): + plate_min_row, plate_min_col, _max_row, _max_col = box + show_labels = len(result.colonies) <= _COLONY_LABEL_MAX_COUNT + + for colony in result.colonies: + min_r_mm, min_c_mm, max_r_mm, max_c_mm = colony.bbox_mm + r0 = min_r_mm / mm_per_px + plate_min_row + c0 = min_c_mm / mm_per_px + plate_min_col + r1 = max_r_mm / mm_per_px + plate_min_row + c1 = max_c_mm / mm_per_px + plate_min_col + + rect = Rectangle( + (c0, r0), + c1 - c0, + r1 - r0, + linewidth=_COLONY_BBOX_THICKNESS, + edgecolor=bbox_color, + facecolor="none", + ) + ax.add_patch(rect) + if show_labels: + ax.text( + (c0 + c1) / 2, + r0 - 4, + str(colony.label), + color=bbox_color, + fontsize=_COLONY_LABEL_FONT_SIZE, + fontweight="bold", + horizontalalignment="center", + verticalalignment="bottom", + clip_on=True, + ) + + fig.subplots_adjust(left=0, right=1, top=1, bottom=0) + canvas.draw() + buf: NDArray[np.uint8] = np.asarray(canvas.buffer_rgba())[:, :, :3].copy() + return buf + + def crop_plates(self) -> list[NDArray[np.uint8]]: + """Return RGB uint8 crops for each detected plate. + + Must be called after :meth:`export_jpg` (or :meth:`detect_plates`) + so that ``plate_boxes`` is populated. + """ + if self.plate_boxes is None: + raise RuntimeError("Call export_jpg() or detect_plates() first.") + img = self._get_rgb_uint8() + crops: list[NDArray[np.uint8]] = [] + for min_row, min_col, max_row, max_col in self.plate_boxes: + crops.append(img[min_row:max_row, min_col:max_col].copy()) + return crops + # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ + def _get_rgb_uint8(self) -> NDArray[np.uint8]: + """Return the cached RGB uint8 version of the loaded intensities.""" + if self._rgb_cache is None: + self._rgb_cache = self._to_rgb_uint8(self.intensities) + return self._rgb_cache + @staticmethod def _to_rgb_uint8(img: NDArray[Any]) -> NDArray[np.uint8]: """Normalize to 8-bit RGB regardless of input dtype/channels.""" - if img.dtype != np.uint8: + orig_dtype = img.dtype + + if img.ndim == 3 and img.shape[2] == 4: + img = ski.color.rgba2rgb(img) + + if orig_dtype != np.uint8: img = ski.util.img_as_ubyte(ski.exposure.rescale_intensity(img)) + elif img.dtype != np.uint8: + # rgba2rgb converted uint8 -> float; scale back without rescaling + img = ski.util.img_as_ubyte(img) if img.ndim == 2: img = ski.color.gray2rgb(img) - elif img.ndim == 3 and img.shape[2] == 4: - img = ski.color.rgba2rgb(img) - img = ski.util.img_as_ubyte(img) return img # type: ignore[return-value] diff --git a/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py b/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py index ba157e67..10c36c3f 100644 --- a/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py +++ b/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py @@ -2,7 +2,11 @@ import logging from data_hub_lambda.api_client import get_client -from data_hub_lambda.epson_v700_scanner.image_processing import TIFFToJPEGConverter +from data_hub_lambda.epson_v700_scanner.colony_detection import ( + export_colony_csv, + run_colony_pipeline, +) +from data_hub_lambda.epson_v700_scanner.image_processing import TiffProcessor from data_hub_shared import s3_utils from data_hub_shared.config import config from data_hub_shared.enums import Instrument @@ -48,9 +52,18 @@ def process_file(run_id: str, filename: str) -> None: s3_utils.download_file(f"s3://{s3_bucket}/{s3_key}", local_file_path) logger.info("Downloaded %s to %s", filename, local_file_path) - converter = TIFFToJPEGConverter(local_file_path) - converter.load() - jpg_file_path = converter.export_jpg() + processor = TiffProcessor(local_file_path) + processor.load() + processor.detect_plates() + + pipeline = None + plate_crops = processor.crop_plates() + if plate_crops: + pipeline = run_colony_pipeline(plate_crops, dpi=processor.dpi) + + jpg_file_path = processor.export_jpg( + colony_results=pipeline.results if pipeline else None, + ) processed_bucket = config.AWS_S3_PROCESSED_DATA_BUCKET jpg_s3_key = f"{INSTRUMENT_ID}/{run_id}/{jpg_file_path.name}" @@ -71,7 +84,32 @@ def process_file(run_id: str, filename: str) -> None: content_type="image/jpeg", ) - metadata = converter.parse_metadata() + metadata = processor.parse_metadata() + + if pipeline: + csv_name = f"{processor.path.stem}_colonies.csv" + csv_path = export_colony_csv( + pipeline.to_dataframes(), + raw_data_dir / csv_name, + ) + csv_s3_key = f"{INSTRUMENT_ID}/{run_id}/{csv_name}" + s3_utils.upload_file(csv_path, f"s3://{processed_bucket}/{csv_s3_key}") + csv_file = client.create_file( + instrument_id=INSTRUMENT_ID, + run_id=run_id, + s3_bucket=processed_bucket or "", + s3_key=csv_s3_key, + filename=csv_name, + category="processed", + ) + client.update_file( + csv_file.id, + size_bytes=csv_path.stat().st_size, + content_type="text/csv", + ) + + metadata["colony_detection"] = pipeline.summaries + logger.info("Parsed metadata: %s", metadata) client.update_run(INSTRUMENT_ID, run_id, metadata=metadata) diff --git a/lambda/tests/epson_v700_scanner/test_colony_detection.py b/lambda/tests/epson_v700_scanner/test_colony_detection.py new file mode 100644 index 00000000..f486fa46 --- /dev/null +++ b/lambda/tests/epson_v700_scanner/test_colony_detection.py @@ -0,0 +1,401 @@ +"""Unit tests for `epson_v700_scanner.colony_detection`.""" + +from __future__ import annotations +from pathlib import Path + +import numpy as np +import pandas as pd + +from data_hub_lambda.epson_v700_scanner.colony_detection import ( + MARGIN_PX, + ColonyDetectionResult, + ColonyProperties, + crop_margin, + detect_colonies, + detect_colony_presence, + export_colony_csv, + measure_colonies, + optimize_colony_contrast, + smooth, + threshold_colonies, +) + + +def _uniform_plate(h: int = 1000, w: int = 1000, value: int = 120) -> np.ndarray: # type: ignore[type-arg] + """Create a uniform-colour plate image (no colonies).""" + return np.full((h, w, 3), value, dtype=np.uint8) + + +def _plate_with_colonies( + h: int = 1000, + w: int = 1000, + bg: int = 120, + colony_color: tuple[int, int, int] = (255, 255, 255), + n_colonies: int = 3, + colony_radius: int = 15, +) -> np.ndarray: # type: ignore[type-arg] + """Create a synthetic plate with circular colonies inside the margin.""" + img = np.full((h, w, 3), bg, dtype=np.uint8) + rng = np.random.RandomState(42) + margin = MARGIN_PX + colony_radius + 5 + for _ in range(n_colonies): + cy = rng.randint(margin, h - margin) + cx = rng.randint(margin, w - margin) + rr, cc = _disk(cy, cx, colony_radius, h, w) + img[rr, cc] = colony_color + return img + + +def _disk(cy: int, cx: int, radius: int, h: int, w: int) -> tuple[np.ndarray, np.ndarray]: # type: ignore[type-arg] + """Return row, col arrays for a filled circle clipped to (h, w).""" + Y, X = np.ogrid[:h, :w] + mask = (Y - cy) ** 2 + (X - cx) ** 2 <= radius**2 + rows, cols = np.where(mask) + return rows, cols + + +# ------------------------------------------------------------------ +# crop_margin +# ------------------------------------------------------------------ + + +class TestCropMargin: + def test_default_removes_margin(self) -> None: + img = np.zeros((1000, 1200, 3), dtype=np.uint8) + cropped = crop_margin(img) + assert cropped.shape == (1000 - 2 * MARGIN_PX, 1200 - 2 * MARGIN_PX, 3) + + def test_grayscale(self) -> None: + img = np.zeros((1000, 1000), dtype=np.uint8) + cropped = crop_margin(img) + assert cropped.shape == (1000 - 2 * MARGIN_PX, 1000 - 2 * MARGIN_PX) + + def test_custom_margin(self) -> None: + img = np.zeros((200, 200, 3), dtype=np.uint8) + cropped = crop_margin(img, margin_px=50) + assert cropped.shape == (100, 100, 3) + + def test_preserves_content(self) -> None: + img = np.arange(500 * 500, dtype=np.uint8).reshape(500, 500) + cropped = crop_margin(img, margin_px=20) + np.testing.assert_array_equal(cropped, img[20:480, 20:480]) + + +# ------------------------------------------------------------------ +# optimize_colony_contrast +# ------------------------------------------------------------------ + + +class TestOptimizeColonyContrast: + def test_grayscale_passthrough(self) -> None: + img = np.random.randint(0, 255, (50, 50), dtype=np.uint8) + result = optimize_colony_contrast(img) + assert result.shape == (50, 50) + assert result.dtype == np.float64 + + def test_uniform_rgb_gives_near_zero(self) -> None: + img = np.full((50, 50, 3), 128, dtype=np.uint8) + result = optimize_colony_contrast(img) + assert result.max() < 1.0 + + def test_colony_pixels_are_bright(self) -> None: + img = np.full((100, 100, 3), 80, dtype=np.uint8) + img[40:60, 40:60] = [220, 220, 220] + result = optimize_colony_contrast(img) + colony_region = result[40:60, 40:60] + bg_region = result[0:20, 0:20] + assert colony_region.mean() > bg_region.mean() + + def test_zero_pixels_stay_zero(self) -> None: + img = np.full((50, 50, 3), 100, dtype=np.uint8) + img[0:10, :] = 0 + result = optimize_colony_contrast(img) + np.testing.assert_array_equal(result[0:10, :], 0) + + +# ------------------------------------------------------------------ +# detect_colony_presence +# ------------------------------------------------------------------ + + +class TestDetectColonyPresence: + def test_uniform_returns_false(self) -> None: + contrast = np.full((100, 100), 1.0) + assert detect_colony_presence(contrast) is False + + def test_high_contrast_returns_true(self) -> None: + contrast = np.zeros((100, 100)) + contrast[30:70, 30:70] = 50.0 + assert detect_colony_presence(contrast) is True + + def test_custom_threshold(self) -> None: + contrast = np.full((100, 100), 3.0) + assert detect_colony_presence(contrast, threshold=2.0) is True + assert detect_colony_presence(contrast, threshold=4.0) is False + + +# ------------------------------------------------------------------ +# smooth +# ------------------------------------------------------------------ + + +class TestSmooth: + def test_output_shape_preserved(self) -> None: + img = np.random.rand(100, 100) + result = smooth(img) + assert result.shape == img.shape + + def test_smoothing_reduces_noise(self) -> None: + rng = np.random.RandomState(0) + img = rng.rand(200, 200) * 100 + result = smooth(img, low_sigma=0.6, high_sigma=64.0) + assert result.std() < img.std() + + def test_result_is_non_negative(self) -> None: + rng = np.random.RandomState(42) + img = rng.rand(200, 200) * 100 + result = smooth(img) + assert np.all(result >= 0) + + +# ------------------------------------------------------------------ +# threshold_colonies +# ------------------------------------------------------------------ + + +class TestThresholdColonies: + def test_returns_bool_mask(self) -> None: + img = np.random.rand(100, 100) + mask = threshold_colonies(img) + assert mask.dtype == np.bool_ + assert mask.shape == img.shape + + def test_bimodal_separation(self) -> None: + img = np.zeros((100, 100), dtype=np.float64) + img[30:70, 30:70] = 100.0 + mask = threshold_colonies(img) + assert mask[50, 50] + assert not mask[0, 0] + + +# ------------------------------------------------------------------ +# measure_colonies +# ------------------------------------------------------------------ + + +_TEST_DPI = 1200 + + +class TestMeasureColonies: + def test_empty_mask_returns_empty(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + rgb = np.zeros((100, 100, 3), dtype=np.uint8) + colonies, cleaned = measure_colonies(mask, rgb, dpi=_TEST_DPI) + assert colonies == [] + assert not cleaned.any() + + def test_single_blob(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + mask[40:60, 40:60] = True + rgb = np.full((100, 100, 3), 128, dtype=np.uint8) + colonies, _ = measure_colonies(mask, rgb, dpi=_TEST_DPI) + assert len(colonies) == 1 + mm_per_px = 25.4 / _TEST_DPI + expected_area_mm2 = 20 * 20 * mm_per_px**2 + assert abs(colonies[0].area_mm2 - expected_area_mm2) < 1e-6 + + def test_two_blobs_sorted_by_area(self) -> None: + mask = np.zeros((200, 200), dtype=bool) + mask[10:22, 10:22] = True # 144 px (above min-area threshold at 1200 DPI) + mask[50:80, 50:80] = True # 900 px + rgb = np.full((200, 200, 3), 128, dtype=np.uint8) + colonies, _ = measure_colonies(mask, rgb, dpi=_TEST_DPI) + assert len(colonies) == 2 + assert colonies[0].area_mm2 > colonies[1].area_mm2 + + def test_colony_properties_populated(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + mask[40:60, 40:60] = True + rgb = np.full((100, 100, 3), 128, dtype=np.uint8) + colonies, _ = measure_colonies(mask, rgb, dpi=_TEST_DPI) + colony = colonies[0] + assert isinstance(colony, ColonyProperties) + assert colony.label >= 1 + assert colony.eccentricity >= 0.0 + assert colony.equivalent_diameter_mm > 0.0 + assert len(colony.bbox_mm) == 4 + + def test_centroid_includes_margin_offset(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + mask[40:60, 40:60] = True + rgb = np.full((100, 100, 3), 128, dtype=np.uint8) + mm_per_px = 25.4 / _TEST_DPI + colonies, _ = measure_colonies(mask, rgb, dpi=_TEST_DPI, margin_px=MARGIN_PX) + colony = colonies[0] + expected_row_mm = (50.0 + MARGIN_PX) * mm_per_px + expected_col_mm = (50.0 + MARGIN_PX) * mm_per_px + assert abs(colony.centroid_row_mm - expected_row_mm) < 0.1 + assert abs(colony.centroid_col_mm - expected_col_mm) < 0.1 + + def test_mean_rgb_populated(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + mask[40:60, 40:60] = True + rgb = np.zeros((100, 100, 3), dtype=np.uint8) + rgb[40:60, 40:60] = [200, 100, 50] + colonies, _ = measure_colonies(mask, rgb, dpi=_TEST_DPI) + colony = colonies[0] + assert abs(colony.mean_rgb[0] - 200.0) < 1.0 + assert abs(colony.mean_rgb[1] - 100.0) < 1.0 + assert abs(colony.mean_rgb[2] - 50.0) < 1.0 + + def test_subthreshold_colony_removed(self) -> None: + """Objects smaller than _MIN_COLONY_AREA_MM2 should be discarded.""" + mask = np.zeros((200, 200), dtype=bool) + # Small blob: 5x5 = 25 px² (well below 112 px threshold at 1200 DPI) + mask[90:95, 90:95] = True + # Large blob: 20x25 = 500 px² (well above threshold) + mask[30:50, 30:55] = True + rgb = np.full((200, 200, 3), 128, dtype=np.uint8) + colonies, cleaned = measure_colonies(mask, rgb, dpi=_TEST_DPI) + assert len(colonies) == 1 + assert colonies[0].area_mm2 > 0.05 + assert not cleaned[90:95, 90:95].any() + + def test_border_colonies_excluded(self) -> None: + mask = np.zeros((100, 100), dtype=bool) + mask[0:10, 40:60] = True # touches top border + mask[40:60, 40:60] = True # interior + rgb = np.full((100, 100, 3), 128, dtype=np.uint8) + colonies, cleaned = measure_colonies(mask, rgb, dpi=_TEST_DPI) + assert len(colonies) == 1 + assert not cleaned[0:10, 40:60].any() + assert cleaned[40:60, 40:60].all() + + +# ------------------------------------------------------------------ +# detect_colonies (full pipeline) +# ------------------------------------------------------------------ + + +class TestDetectColonies: + def test_uniform_plate_no_colonies(self) -> None: + plate = _uniform_plate() + result = detect_colonies(plate, dpi=_TEST_DPI) + assert isinstance(result, ColonyDetectionResult) + assert result.has_colonies is False + assert result.colonies == [] + assert result.mask.shape == result.contrast.shape + assert not result.mask.any() + + def test_plate_with_colonies_detected(self) -> None: + plate = _plate_with_colonies(colony_color=(255, 255, 255), n_colonies=8, colony_radius=30) + result = detect_colonies(plate, dpi=_TEST_DPI) + assert result.has_colonies is True + assert len(result.colonies) > 0 + + def test_summary_schema(self) -> None: + plate = _plate_with_colonies(n_colonies=2) + result = detect_colonies(plate, dpi=_TEST_DPI) + summary = result.summary() + assert "has_colonies" in summary + assert "colony_count" in summary + assert isinstance(summary["colonies"], list) + if summary["colonies"]: + c = summary["colonies"][0] + assert "label" in c + assert "area_mm2" in c + assert "centroid_mm" in c + assert "bbox_mm" in c + assert "eccentricity" in c + assert "equivalent_diameter_mm" in c + assert "mean_rgb" in c + + def test_cropped_smaller_than_input(self) -> None: + plate = _uniform_plate(1000, 1000) + result = detect_colonies(plate, dpi=_TEST_DPI) + assert result.cropped.shape[0] < 1000 + assert result.cropped.shape[1] < 1000 + + def test_grayscale_input(self) -> None: + plate = np.full((1000, 1000), 128, dtype=np.uint8) + result = detect_colonies(plate, dpi=_TEST_DPI) + assert isinstance(result, ColonyDetectionResult) + + +# ------------------------------------------------------------------ +# to_dataframe +# ------------------------------------------------------------------ + + +class TestToDataframe: + def test_columns_present(self) -> None: + plate = _plate_with_colonies(colony_color=(255, 255, 255), n_colonies=8, colony_radius=30) + result = detect_colonies(plate, dpi=_TEST_DPI) + df = result.to_dataframe(plate_index=1) + assert isinstance(df, pd.DataFrame) + expected_cols = { + "plate_index", + "label", + "area_mm2", + "centroid_row_mm", + "centroid_col_mm", + "bbox_min_row_mm", + "bbox_min_col_mm", + "bbox_max_row_mm", + "bbox_max_col_mm", + "eccentricity", + "equivalent_diameter_mm", + "mean_r", + "mean_g", + "mean_b", + } + assert expected_cols == set(df.columns) + + def test_row_count_matches_colonies(self) -> None: + plate = _plate_with_colonies(colony_color=(255, 255, 255), n_colonies=8, colony_radius=30) + result = detect_colonies(plate, dpi=_TEST_DPI) + df = result.to_dataframe() + assert len(df) == len(result.colonies) + + def test_empty_result_gives_empty_df(self) -> None: + plate = _uniform_plate() + result = detect_colonies(plate, dpi=_TEST_DPI) + df = result.to_dataframe() + assert len(df) == 0 + assert "plate_index" in df.columns + + def test_plate_index_propagated(self) -> None: + plate = _plate_with_colonies(colony_color=(255, 255, 255), n_colonies=8, colony_radius=30) + result = detect_colonies(plate, dpi=_TEST_DPI) + df = result.to_dataframe(plate_index=3) + assert (df["plate_index"] == 3).all() + + +# ------------------------------------------------------------------ +# export_colony_csv +# ------------------------------------------------------------------ + + +class TestExportColonyCsv: + def test_writes_csv(self, tmp_path: Path) -> None: + plate = _plate_with_colonies(colony_color=(255, 255, 255), n_colonies=8, colony_radius=30) + result = detect_colonies(plate, dpi=_TEST_DPI) + frames = [result.to_dataframe(plate_index=1)] + csv_path = export_colony_csv(frames, tmp_path / "colonies.csv") + assert csv_path.exists() + loaded = pd.read_csv(csv_path) + assert len(loaded) == len(result.colonies) + + def test_concatenates_multiple_plates(self, tmp_path: Path) -> None: + df1 = pd.DataFrame({"plate_index": [1], "label": [1], "area_mm2": [0.5]}) + df2 = pd.DataFrame({"plate_index": [2], "label": [1], "area_mm2": [1.0]}) + csv_path = export_colony_csv([df1, df2], tmp_path / "multi.csv") + loaded = pd.read_csv(csv_path) + assert len(loaded) == 2 + assert list(loaded["plate_index"]) == [1, 2] + + def test_empty_frames_writes_empty_csv(self, tmp_path: Path) -> None: + csv_path = export_colony_csv([], tmp_path / "empty.csv") + assert csv_path.exists() + loaded = pd.read_csv(csv_path) + assert len(loaded) == 0 diff --git a/lambda/tests/epson_v700_scanner/test_image_processing.py b/lambda/tests/epson_v700_scanner/test_image_processing.py index b76c8c76..0bd1e550 100644 --- a/lambda/tests/epson_v700_scanner/test_image_processing.py +++ b/lambda/tests/epson_v700_scanner/test_image_processing.py @@ -3,18 +3,20 @@ from __future__ import annotations from pathlib import Path +import imageio.v3 as iio import numpy as np import pytest -import skimage as ski import tifffile from data_hub_lambda.epson_v700_scanner.image_processing import ( MAX_DIMENSION, - TIFFToJPEGConverter, + TiffProcessor, _derive_color_mode, _derive_dpi, ) +_GOLD_RGB = np.array([200, 170, 40], dtype=np.uint8) + def _write_tiff(path: Path, img: np.ndarray) -> Path: # type: ignore[type-arg] tifffile.imwrite(str(path), img) @@ -26,14 +28,14 @@ def test_produces_valid_jpeg(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (200, 300, 3), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "scan.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() assert jpg_path.exists() assert jpg_path.suffix == ".jpg" - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert loaded.ndim == 3 assert loaded.shape[2] == 3 @@ -41,22 +43,22 @@ def test_resizes_large_image(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (4000, 6000, 3), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "big.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert max(loaded.shape[:2]) <= MAX_DIMENSION def test_does_not_upscale_small_image(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (100, 150, 3), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "small.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert loaded.shape[0] == 100 assert loaded.shape[1] == 150 @@ -64,11 +66,11 @@ def test_converts_grayscale_to_rgb(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (200, 300), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "gray.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert loaded.ndim == 3 assert loaded.shape[2] == 3 @@ -76,11 +78,11 @@ def test_converts_rgba_to_rgb(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (200, 300, 4), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "rgba.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert loaded.ndim == 3 assert loaded.shape[2] == 3 @@ -88,11 +90,11 @@ def test_handles_16bit_input(self, tmp_path: Path) -> None: img = np.random.randint(0, 65535, (200, 300, 3), dtype=np.uint16) tif_path = _write_tiff(tmp_path / "16bit.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() jpg_path = converter.export_jpg() - loaded = ski.io.imread(str(jpg_path)) + loaded = iio.imread(jpg_path) assert loaded.dtype == np.uint8 @@ -101,7 +103,7 @@ def test_returns_original_dimensions(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "scan.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() metadata = converter.parse_metadata() @@ -112,7 +114,7 @@ def test_returns_standard_tiff_tags(self, tmp_path: Path) -> None: img = np.random.randint(0, 255, (100, 200, 3), dtype=np.uint8) tif_path = _write_tiff(tmp_path / "scan.tif", img) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() metadata = converter.parse_metadata() @@ -166,7 +168,7 @@ def test_emits_dpi_and_rgb_color_mode(self, tmp_path: Path) -> None: tif_path = tmp_path / "scan.tif" tifffile.imwrite(str(tif_path), img, resolution=(300, 300)) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() metadata = converter.parse_metadata() @@ -178,7 +180,7 @@ def test_emits_bw_color_mode_for_grayscale(self, tmp_path: Path) -> None: tif_path = tmp_path / "gray.tif" tifffile.imwrite(str(tif_path), img, resolution=(600, 600)) - converter = TIFFToJPEGConverter(tif_path) + converter = TiffProcessor(tif_path) converter.load() metadata = converter.parse_metadata() @@ -188,13 +190,135 @@ def test_emits_bw_color_mode_for_grayscale(self, tmp_path: Path) -> None: class TestValidation: def test_rejects_missing_file(self, tmp_path: Path) -> None: - converter = TIFFToJPEGConverter(tmp_path / "nonexistent.tif") + converter = TiffProcessor(tmp_path / "nonexistent.tif") with pytest.raises(FileNotFoundError): converter.load() def test_rejects_non_tiff_extension(self, tmp_path: Path) -> None: path = tmp_path / "scan.png" path.write_bytes(b"fake") - converter = TIFFToJPEGConverter(path) + converter = TiffProcessor(path) with pytest.raises(ValueError, match="Expected TIFF file"): converter.load() + + +# ------------------------------------------------------------------ +# Plate detection +# ------------------------------------------------------------------ + + +def _make_gold_frame( + canvas_h: int, + canvas_w: int, + top: int, + left: int, + bottom: int, + right: int, + thickness: int = 20, +) -> np.ndarray: # type: ignore[type-arg] + """Paint a gold rectangular frame on a black canvas and return it.""" + img = np.zeros((canvas_h, canvas_w, 3), dtype=np.uint8) + img[top : top + thickness, left:right] = _GOLD_RGB + img[bottom - thickness : bottom, left:right] = _GOLD_RGB + img[top:bottom, left : left + thickness] = _GOLD_RGB + img[top:bottom, right - thickness : right] = _GOLD_RGB + return img + + +class TestDetectPlates: + def test_single_gold_frame(self) -> None: + img = _make_gold_frame(500, 400, top=50, left=50, bottom=350, right=350) + proc = TiffProcessor(Path("dummy.tif")) + proc.detect_plates(img) + + assert proc.plate_boxes is not None + assert len(proc.plate_boxes) == 1 + min_row, min_col, max_row, max_col = proc.plate_boxes[0] + assert 60 < min_row < 80 + assert 60 < min_col < 80 + assert 320 < max_row < 340 + assert 320 < max_col < 340 + + def test_two_gold_frames_sorted_left_to_right(self) -> None: + img = np.zeros((500, 800, 3), dtype=np.uint8) + frame1 = _make_gold_frame(500, 800, top=50, left=50, bottom=350, right=300) + frame2 = _make_gold_frame(500, 800, top=50, left=400, bottom=350, right=700) + img = np.maximum(img, np.maximum(frame1, frame2)) + + proc = TiffProcessor(Path("dummy.tif")) + proc.detect_plates(img) + + assert proc.plate_boxes is not None + assert len(proc.plate_boxes) == 2 + assert proc.plate_boxes[0][1] < proc.plate_boxes[1][1] + + def test_no_gold_returns_empty(self) -> None: + img = np.random.randint(0, 50, (400, 400, 3), dtype=np.uint8) + proc = TiffProcessor(Path("dummy.tif")) + proc.detect_plates(img) + assert proc.plate_boxes == [] + + def test_metadata_includes_plate_count(self, tmp_path: Path) -> None: + img = _make_gold_frame(500, 400, top=50, left=50, bottom=350, right=350) + tif_path = tmp_path / "framed.tif" + tifffile.imwrite(str(tif_path), img) + + proc = TiffProcessor(tif_path) + proc.load() + proc.export_jpg() + metadata = proc.parse_metadata() + + assert metadata["plate_count"] == 1 + assert len(metadata["plate_boxes"]) == 1 + + +class TestDrawPlateOverlays: + def test_shape_and_dtype_preserved(self) -> None: + img = np.random.randint(0, 255, (200, 300, 3), dtype=np.uint8) + boxes = [(50, 50, 150, 250)] + result = TiffProcessor._draw_plate_overlays(img, boxes) + assert result.shape == img.shape + assert result.dtype == np.uint8 + + def test_interior_pixels_unchanged(self) -> None: + img = np.full((200, 300, 3), 128, dtype=np.uint8) + boxes = [(50, 50, 150, 250)] + result = TiffProcessor._draw_plate_overlays(img, boxes) + np.testing.assert_array_equal(result[50:150, 50:250], img[50:150, 50:250]) + + def test_border_pixels_modified(self) -> None: + img = np.full((200, 300, 3), 128, dtype=np.uint8) + boxes = [(50, 50, 150, 250)] + result = TiffProcessor._draw_plate_overlays(img, boxes) + assert not np.array_equal(result[42:50, 50:250], img[42:50, 50:250]) + + +class TestDrawColonyBboxes: + def test_shape_and_dtype_preserved(self) -> None: + from data_hub_lambda.epson_v700_scanner.colony_detection import ( + ColonyDetectionResult, + ColonyProperties, + ) + + img = np.random.randint(0, 255, (400, 400, 3), dtype=np.uint8) + boxes = [(50, 50, 350, 350)] + colony = ColonyProperties( + label=1, + area_mm2=1.0, + centroid_row_mm=5.0, + centroid_col_mm=5.0, + bbox_mm=(4.0, 4.0, 6.0, 6.0), + eccentricity=0.0, + equivalent_diameter_mm=1.13, + mean_rgb=(128.0, 128.0, 128.0), + ) + result_obj = ColonyDetectionResult( + cropped=img[50:350, 50:350], + contrast=np.zeros((300, 300)), + has_colonies=True, + mask=np.zeros((300, 300), dtype=bool), + colonies=[colony], + ) + result = TiffProcessor._draw_colony_bboxes(img, boxes, [result_obj], dpi=600) + assert result.shape == img.shape + assert result.dtype == np.uint8 diff --git a/lambda/tests/epson_v700_scanner/test_process_file.py b/lambda/tests/epson_v700_scanner/test_process_file.py index cd08522d..cbdc80cd 100644 --- a/lambda/tests/epson_v700_scanner/test_process_file.py +++ b/lambda/tests/epson_v700_scanner/test_process_file.py @@ -1,7 +1,7 @@ """Unit tests for `epson_v700_scanner.process_file`. These verify the orchestration logic by mocking S3 I/O, the API client, -and the TIFFToJPEGConverter. +and the TiffProcessor. """ from __future__ import annotations @@ -69,7 +69,7 @@ def patched_jpg_path(tmp_path: Path) -> Path: @pytest.fixture def patched_converter(patched_jpg_path: Path) -> MagicMock: - """A stand-in TIFFToJPEGConverter whose export_jpg returns a real tmp file.""" + """A stand-in TiffProcessor whose export_jpg returns a real tmp file.""" converter = MagicMock() converter.load.return_value = None converter.export_jpg.return_value = patched_jpg_path @@ -80,6 +80,7 @@ def patched_converter(patched_jpg_path: Path) -> MagicMock: "OriginalHeight": 4800, "OriginalWidth": 6400, } + converter.crop_plates.return_value = [] return converter @@ -97,7 +98,7 @@ def test_creates_run_and_files( patch(f"{_PATCH_PREFIX}.get_client", return_value=client), patch(f"{_PATCH_PREFIX}.s3_utils") as s3_mock, patch( - f"{_PATCH_PREFIX}.TIFFToJPEGConverter", + f"{_PATCH_PREFIX}.TiffProcessor", return_value=patched_converter, ), ): @@ -120,7 +121,7 @@ def test_uploads_jpg_and_registers_processed_file( patch(f"{_PATCH_PREFIX}.get_client", return_value=client), patch(f"{_PATCH_PREFIX}.s3_utils"), patch( - f"{_PATCH_PREFIX}.TIFFToJPEGConverter", + f"{_PATCH_PREFIX}.TiffProcessor", return_value=patched_converter, ), ): @@ -144,7 +145,7 @@ def test_stores_metadata_on_run( patch(f"{_PATCH_PREFIX}.get_client", return_value=client), patch(f"{_PATCH_PREFIX}.s3_utils"), patch( - f"{_PATCH_PREFIX}.TIFFToJPEGConverter", + f"{_PATCH_PREFIX}.TiffProcessor", return_value=patched_converter, ), ): @@ -166,7 +167,7 @@ def test_marks_raw_file_completed( patch(f"{_PATCH_PREFIX}.get_client", return_value=client), patch(f"{_PATCH_PREFIX}.s3_utils"), patch( - f"{_PATCH_PREFIX}.TIFFToJPEGConverter", + f"{_PATCH_PREFIX}.TiffProcessor", return_value=patched_converter, ), ): @@ -193,7 +194,7 @@ def test_marks_raw_file_failed_on_exception(self) -> None: patch(f"{_PATCH_PREFIX}.get_client", return_value=client), patch(f"{_PATCH_PREFIX}.s3_utils"), patch( - f"{_PATCH_PREFIX}.TIFFToJPEGConverter", + f"{_PATCH_PREFIX}.TiffProcessor", return_value=failing_converter, ), ):