Complete Example Analysis
This page provides a complete, working example of a fruit bruise detection analysis module. This example demonstrates all key concepts needed to create a custom analysis for Granny.
Overview
This example implements a BruiseDetection analysis that:
Detects bruises on fruit images using color thresholding
Calculates bruise percentage
Provides adjustable threshold and visualization parameters
Outputs annotated images and CSV results
Uses the multiprocessing architecture via
_preRun(),_processImage(),_postRun()
The algorithm uses LAB color space to detect darker regions that indicate bruising.
Complete Implementation
File: Granny/Analyses/BruiseDetection.py
"""
Bruise detection analysis for fruit quality assessment.
This module detects and quantifies bruises on fruit images by analyzing
color differences in LAB color space. Darker regions below a threshold
are classified as bruises.
The analysis outputs:
- Annotated images showing bruise regions
- CSV file with bruise percentages for each fruit
- Metadata including threshold values and timestamp
Author: Example Author
Date: 2024-01-15
"""
import csv
import os
from datetime import datetime
from typing import Dict, List, Tuple
import cv2
import numpy as np
from numpy.typing import NDArray
from Granny.Analyses.Analysis import Analysis
from Granny.Models.Images.Image import Image
from Granny.Models.IO.RGBImageFile import RGBImageFile
from Granny.Models.Values.IntValue import IntValue
from Granny.Models.Values.FloatValue import FloatValue
from Granny.Models.Values.StringValue import StringValue
from Granny.Models.Values.ImageListValue import ImageListValue
from Granny.Models.Values.MetaDataValue import MetaDataValue
class BruiseDetection(Analysis):
"""
Analysis class for detecting and quantifying bruises on fruit images.
This analysis processes individual fruit images and detects bruises by
identifying darker regions in LAB color space. Results include both
visual annotations and quantitative measurements.
The base Analysis class handles:
- Loading images from input directory
- Parallel processing via multiprocessing.Pool
- CPU core management
You implement:
- _preRun(): Setup before processing
- _processImage(): Process single image (runs in parallel)
- _postRun(): Save results after all images processed
Attributes:
images (List[Image]): Loaded input images (set by base class)
input_images (ImageListValue): Input directory parameter
output_images (ImageListValue): Output directory for annotated images
lightness_threshold (IntValue): L-channel threshold for bruise detection
min_bruise_area (IntValue): Minimum area (pixels) for valid bruise
morphological_kernel (IntValue): Kernel size for noise removal
mask_alpha (FloatValue): Transparency for bruise mask overlay
"""
# This name will be used in CLI: granny -i cli --analysis bruise
__analysis_name__ = "bruise"
def __init__(self):
"""
Initialize the bruise detection analysis with default parameters.
Sets up input/output directories, detection thresholds, and
visualization parameters with sensible defaults.
"""
# STEP 1: Initialize base class (REQUIRED)
super().__init__()
# STEP 2: Set up input parameter
self.input_images = ImageListValue(
"input",
"input",
"Directory containing fruit images to analyze for bruises."
)
self.input_images.setIsRequired(True) # User must provide this
self.addInParam(self.input_images)
# STEP 3: Set up output directories
# Create timestamped output directory
result_dir = os.path.join(
os.curdir,
"results",
self.__analysis_name__,
datetime.now().strftime("%Y-%m-%d-%H-%M")
)
self.output_images = ImageListValue(
"output",
"output",
"Directory where annotated images will be saved."
)
self.output_images.setValue(result_dir)
self.addInParam(self.output_images)
self.output_results = MetaDataValue(
"results",
"results",
"Directory where CSV results will be saved."
)
self.output_results.setValue(result_dir)
# STEP 4: Define analysis parameters
# These parameters affect the analysis results
self.lightness_threshold = IntValue(
"l_thresh",
"lightness_threshold",
"Lightness (L channel in LAB) threshold for bruise detection. "
"Pixels with L values below this are considered bruises. "
"Range: 0-255, default: 80."
)
self.lightness_threshold.setMin(0)
self.lightness_threshold.setMax(255)
self.lightness_threshold.setValue(80)
self.lightness_threshold.setIsRequired(False)
self.addInParam(self.lightness_threshold)
self.min_bruise_area = IntValue(
"min_area",
"min_bruise_area",
"Minimum area in pixels for a region to be counted as a bruise. "
"Smaller regions are considered noise. "
"Range: 1-10000, default: 100."
)
self.min_bruise_area.setMin(1)
self.min_bruise_area.setMax(10000)
self.min_bruise_area.setValue(100)
self.min_bruise_area.setIsRequired(False)
self.addInParam(self.min_bruise_area)
self.morphological_kernel = IntValue(
"morph_kernel",
"morph_kernel",
"Kernel size for morphological operations (noise removal). "
"Larger values = more smoothing. "
"Range: 1-99, default: 5."
)
self.morphological_kernel.setMin(1)
self.morphological_kernel.setMax(99)
self.morphological_kernel.setValue(5)
self.morphological_kernel.setIsRequired(False)
self.addInParam(self.morphological_kernel)
# STEP 5: Define visualization parameters
# These parameters only affect the visual output, not the measurements
self.mask_alpha = FloatValue(
"alpha",
"mask_alpha",
"Transparency of bruise mask overlay on output images. "
"0.0 = transparent, 1.0 = opaque. "
"Range: 0.0-1.0, default: 0.6."
)
self.mask_alpha.setMin(0.0)
self.mask_alpha.setMax(1.0)
self.mask_alpha.setValue(0.6)
self.mask_alpha.setIsRequired(False)
self.addInParam(self.mask_alpha)
self.font_scale = FloatValue(
"font",
"font_scale",
"Font scale for text annotations on output images. "
"Range: 0.1-10.0, default: 1.0."
)
self.font_scale.setMin(0.1)
self.font_scale.setMax(10.0)
self.font_scale.setValue(1.0)
self.font_scale.setIsRequired(False)
self.addInParam(self.font_scale)
# Bruise mask color (BGR format for OpenCV)
self.bruise_color = (0, 0, 255) # Red
# =========================================================================
# REQUIRED ABSTRACT METHODS
# =========================================================================
def _preRun(self):
"""
Setup before image processing begins.
Called once by performAnalysis() before parallel processing starts.
self.images is already populated with loaded Image objects.
Use this for:
- Getting parameter values
- Initializing output directory
- Printing analysis info
"""
# Get output directory path
self.output_dir = self.output_images.getValue()
self.results_dir = self.output_results.getValue()
# Print analysis info
print(f"\n{'='*60}")
print(f"BRUISE DETECTION ANALYSIS")
print(f"{'='*60}")
print(f"Input directory: {self.input_images.getValue()}")
print(f"Output directory: {self.output_dir}")
print(f"\nAnalysis Parameters:")
print(f" Lightness threshold: {self.lightness_threshold.getValue()}")
print(f" Min bruise area: {self.min_bruise_area.getValue()} pixels")
print(f" Morph kernel: {self.morphological_kernel.getValue()}")
print(f"\nVisualization Parameters:")
print(f" Mask alpha: {self.mask_alpha.getValue()}")
print(f" Font scale: {self.font_scale.getValue()}")
print(f"\nProcessing {len(self.images)} images...")
print(f"{'='*60}\n")
def _processImage(self, image: Image) -> Image:
"""
Process a single image for bruise detection.
This method runs in PARALLEL across multiple CPU cores.
Each call receives one Image and must return the processed Image.
IMPORTANT: Don't modify shared state here - it won't work
with multiprocessing. Return all results via the Image object.
Args:
image: Input Image instance (filepath set, image not loaded)
Returns:
Image: The same Image with processed data and metadata
"""
# Get parameter values (safe - these are read-only)
l_thresh = self.lightness_threshold.getValue()
min_area = self.min_bruise_area.getValue()
kernel_size = self.morphological_kernel.getValue()
alpha = self.mask_alpha.getValue()
font_scale = self.font_scale.getValue()
# Load the image data
image_io = RGBImageFile()
image_io.setFilePath(image.getFilePath())
image.loadImage(image_io)
# Get the numpy array (BGR format from OpenCV)
img_array = image.getImage()
# Perform bruise detection
result_array, bruise_pct, bruise_count = self._detect_bruises(
img_array, l_thresh, min_area, kernel_size, alpha, font_scale
)
# Update the image with processed result
image.setImage(result_array)
# Add metadata to the image (will be collected in _postRun)
bruise_pct_val = FloatValue("bruise_percentage", "bruise_pct", "Bruise percentage")
bruise_pct_val.setValue(bruise_pct)
image.addValue(bruise_pct_val)
bruise_count_val = IntValue("bruise_count", "bruise_count", "Number of bruise regions")
bruise_count_val.setValue(bruise_count)
image.addValue(bruise_count_val)
threshold_val = IntValue("threshold_used", "threshold", "L threshold used")
threshold_val.setValue(l_thresh)
image.addValue(threshold_val)
return image
def _postRun(self, results: List[Image]) -> List[Image]:
"""
Post-processing after all images are processed.
Called once with all processed Image objects.
Use this to save images, generate CSV reports, print summaries.
Args:
results: List of processed Image objects from _processImage()
Returns:
List[Image]: The final list of result images
"""
print(f"\nSaving {len(results)} images to: {self.output_dir}")
# Ensure output directory exists
os.makedirs(self.output_dir, exist_ok=True)
# Save each image and collect CSV data
image_io = RGBImageFile()
csv_data = []
for image in results:
# Save the image
image.saveImage(image_io, self.output_dir)
# Collect data for CSV
metadata = image.getMetaData()
csv_data.append({
"filename": image.getImageName(),
"bruise_percentage": f"{metadata['bruise_percentage'].getValue():.2f}",
"bruise_count": metadata['bruise_count'].getValue(),
"lightness_threshold": metadata['threshold_used'].getValue()
})
print(f" Saved: {image.getImageName()} - "
f"Bruise: {metadata['bruise_percentage'].getValue():.2f}%")
# Save CSV results
self._save_csv(csv_data, self.results_dir)
print(f"\n{'='*60}")
print(f"Analysis complete! Processed {len(results)} images.")
print(f"{'='*60}\n")
return results
# =========================================================================
# HELPER METHODS
# =========================================================================
def _detect_bruises(
self,
img: NDArray,
l_thresh: int,
min_area: int,
kernel_size: int,
alpha: float,
font_scale: float
) -> Tuple[NDArray, float, int]:
"""
Detect bruises on a single fruit image.
Algorithm:
1. Convert image to LAB color space
2. Threshold L channel to find dark regions
3. Apply morphological operations to remove noise
4. Filter small regions
5. Calculate bruise percentage
6. Annotate image with results
Args:
img: Input image as NumPy array (BGR format)
l_thresh: Lightness threshold
min_area: Minimum bruise area in pixels
kernel_size: Morphological kernel size
alpha: Mask transparency
font_scale: Text font scale
Returns:
Tuple of (annotated_image, bruise_percentage, bruise_count)
"""
# Convert to LAB color space
lab = cv2.cvtColor(img, cv2.COLOR_BGR2LAB)
l_channel, a_channel, b_channel = cv2.split(lab)
# Threshold L channel - dark regions are potential bruises
_, binary = cv2.threshold(l_channel, l_thresh, 255, cv2.THRESH_BINARY_INV)
# Morphological operations to remove noise
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (kernel_size, kernel_size))
binary = cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel)
binary = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)
# Find connected components (bruise regions)
num_labels, labels, stats, centroids = cv2.connectedComponentsWithStats(
binary, connectivity=8
)
# Filter small regions and count valid bruises
bruise_mask = np.zeros_like(binary)
bruise_count = 0
total_bruise_area = 0
for i in range(1, num_labels): # Skip background (label 0)
area = stats[i, cv2.CC_STAT_AREA]
if area >= min_area:
bruise_mask[labels == i] = 255
bruise_count += 1
total_bruise_area += area
# Calculate bruise percentage
total_pixels = img.shape[0] * img.shape[1]
bruise_pct = (total_bruise_area / total_pixels) * 100
# Create visualization
annotated = self._create_visualization(
img, bruise_mask, bruise_pct, bruise_count, alpha, font_scale
)
return annotated, bruise_pct, bruise_count
def _create_visualization(
self,
img: NDArray,
bruise_mask: NDArray,
bruise_pct: float,
bruise_count: int,
alpha: float,
font_scale: float
) -> NDArray:
"""
Create annotated visualization of bruise detection results.
Args:
img: Original image
bruise_mask: Binary mask of bruise regions
bruise_pct: Bruise percentage
bruise_count: Number of bruise regions
alpha: Transparency for mask overlay
font_scale: Font scale for text
Returns:
Annotated image
"""
# Create colored mask
mask_color = np.zeros_like(img)
mask_color[bruise_mask == 255] = self.bruise_color
# Blend with original image
result = cv2.addWeighted(img, 1.0, mask_color, alpha, 0)
# Add text annotations
thickness = 2
font = cv2.FONT_HERSHEY_SIMPLEX
# Bruise percentage
text1 = f"Bruise: {bruise_pct:.2f}%"
cv2.putText(result, text1, (20, 50), font, font_scale,
(255, 255, 255), thickness, cv2.LINE_AA)
# Bruise count
text2 = f"Regions: {bruise_count}"
cv2.putText(result, text2, (20, 100), font, font_scale,
(255, 255, 255), thickness, cv2.LINE_AA)
return result
def _save_csv(self, data: List[dict], output_dir: str) -> None:
"""
Save analysis results to CSV file.
Args:
data: List of dictionaries containing results for each image
output_dir: Directory to save CSV file
"""
if not data:
print("Warning: No data to save to CSV.")
return
# Ensure directory exists
os.makedirs(output_dir, exist_ok=True)
# Create CSV filename
csv_path = os.path.join(output_dir, f"{self.__analysis_name__}_results.csv")
# Write CSV
with open(csv_path, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
print(f"Results saved to: {csv_path}")
Integration Steps
After creating the file, follow these steps to integrate it into Granny:
Update CLI Interface
Edit
Granny/Interfaces/UI/GrannyCLI.py:Add import:
from Granny.Analyses.BruiseDetection import BruiseDetection
Update choices:
choices=["segmentation", "blush", "color", "scald", "starch", "bruise"]
Create Test File
Create
tests/test_Analyses/test_BruiseDetection.py:from Granny.Analyses.BruiseDetection import BruiseDetection import pytest def test_init(): """Test analysis initialization.""" analysis = BruiseDetection() assert analysis.__analysis_name__ == "bruise" assert analysis.lightness_threshold.getValue() == 80 assert analysis.min_bruise_area.getValue() == 100 def test_parameters(): """Test parameter constraints.""" analysis = BruiseDetection() # Test threshold bounds assert analysis.lightness_threshold.getMin() == 0 assert analysis.lightness_threshold.getMax() == 255 # Test mask alpha bounds assert analysis.mask_alpha.getMin() == 0.0 assert analysis.mask_alpha.getMax() == 1.0 def test_parameter_setting(): """Test setting parameters.""" analysis = BruiseDetection() analysis.lightness_threshold.setValue(90) assert analysis.lightness_threshold.getValue() == 90 analysis.min_bruise_area.setValue(200) assert analysis.min_bruise_area.getValue() == 200
Usage Examples
Basic usage with defaults:
granny -i cli --analysis bruise --input ./demo/apple_images/
Custom threshold:
granny -i cli --analysis bruise \
--input ./images/ \
--lightness_threshold 75 \
--min_bruise_area 150
Full customization:
granny -i cli --analysis bruise \
--input ./fruit_photos/ \
--output ./my_results/ \
--lightness_threshold 70 \
--min_bruise_area 200 \
--morph_kernel 7 \
--mask_alpha 0.7 \
--font_scale 1.5
Specify CPU cores:
granny -i cli --analysis bruise \
--input ./images/ \
--cpu 4
Running Tests
# Run all tests
python -m pytest tests/test_Analyses/test_BruiseDetection.py -v
# Run specific test
python -m pytest tests/test_Analyses/test_BruiseDetection.py::test_init -v
Key Takeaways
This example demonstrates:
Proper class structure with inheritance from
AnalysisThree abstract methods:
_preRun(),_processImage(),_postRun()Parallel processing via the base class (no manual Pool management)
Type-safe parameters using IntValue, FloatValue, ImageListValue
Parameter constraints with min/max ranges
Comprehensive docstrings for class and methods
Image processing workflow using OpenCV and NumPy
Result storage for both images and CSV data
Metadata handling via
addValue()on Image objectsUser feedback via print statements
Architecture Notes
Why three methods instead of one performAnalysis()?
The base Analysis.performAnalysis() handles:
Loading images from the input directory
Managing the multiprocessing Pool
CPU core allocation (80% default, or user-specified)
Calling your methods in the right order
This means you get parallel processing for free just by implementing the three methods correctly.
Multiprocessing constraints in _processImage():
Since _processImage() runs in separate processes:
Don’t modify instance variables (changes won’t propagate)
Return results via the Image object’s metadata
Aggregation happens in
_postRun()Parameter values can be read (they’re pickled with the object)
Best Practices Demonstrated:
Separation of concerns: Detection logic separate from visualization
Configurable parameters: All magic numbers are parameters
Error handling: Check for empty image lists
Progress feedback: Print statements for user awareness
Consistent naming: Follow existing Granny conventions
Documentation: Complete docstrings and inline comments
Type hints: Use NDArray and type annotations
Timestamped results: Avoid overwriting previous results
CSV output: Provide machine-readable results
Next Steps
Modify this example for your specific use case
Add additional parameters as needed
Implement more sophisticated algorithms
Create unit tests for your specific analysis
Update documentation with your analysis details