Source code for phenotypic.abc_._base_operation
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
pass
import importlib.util
import logging
import tracemalloc
from abc import ABC
# Check for optional dependencies
PYMPLER_AVAILABLE = importlib.util.find_spec("pympler") is not None
PSUTIL_AVAILABLE = importlib.util.find_spec("psutil") is not None
# Import if available
if PYMPLER_AVAILABLE:
from pympler import muppy, summary
if PSUTIL_AVAILABLE:
import psutil
[docs]
class BaseOperation(ABC):
"""Root abstract base class for all operations in PhenoTypic.
BaseOperation is the foundation of PhenoTypic's operation system. It provides
automatic memory tracking, logging integration, and utilities for parallel
execution. All operations in PhenoTypic inherit from BaseOperation (either
directly or through intermediate ABCs like ImageOperation and MeasureFeatures).
This class is a blueprint for extending the framework: when you create a new
operation, BaseOperation automatically handles memory profiling and logging so
you can focus on the algorithm implementation.
What it provides automatically:
- **Memory Tracking:** BaseOperation automatically initiates tracemalloc when
the logger is enabled for INFO level or higher. This enables per-operation
memory usage monitoring without explicit instrumentation. Three levels of
memory tracking are available:
1. Object memory (via pympler if available): Detailed breakdown of memory
used by Python objects in your operation.
2. Process memory (via psutil if available): System-level memory usage
(RSS - resident set size).
3. Tracemalloc snapshots: Python's built-in memory tracking showing current
and peak allocations.
- **Logging Integration:** A logger is created automatically for each operation
class with the name format: `module.ClassName`. Subclasses can log messages
and memory usage without additional setup.
- **Parallel Execution Support:** Operations are serialized with all instance
attributes (`op.__dict__`) for parallel execution. Worker processes unpickle
the complete operation object and execute it.
Inheritance hierarchy:
BaseOperation (this class)
├── ImageOperation
│ ├── ImageEnhancer (preprocessing filters, noise reduction)
│ ├── ImageCorrector (rotation, alignment, quality fixes)
│ └── ObjectDetector (colony detection algorithms)
│
├── MeasureFeatures (feature extraction from detected objects)
│
└── GridOperation (grid detection and refinement)
How to subclass BaseOperation:
When extending BaseOperation, you typically implement one of its subclasses
(ImageOperation, MeasureFeatures, etc.) which provides the specific interface
for your operation type. All the memory tracking and logging happens
automatically in the parent class.
Example: Creating a custom operation (without image details)::
from phenotypic.abc_ import BaseOperation
import logging
class MyCustomOperation(BaseOperation):
def __init__(self, param1, param2=5):
# Always call parent __init__ first
super().__init__()
# Store your parameters as attributes
self.param1 = param1
self.param2 = param2
def _operate(self, data):
# Your algorithm here
# Logger available as self._logger
self._logger.info(f"Processing with param1={self.param1}")
# Log memory usage after expensive operations
self._log_memory_usage("after processing")
return result
Attributes:
_logger (logging.Logger): Logger instance created automatically with
the format `module.ClassName`. Use `_logger.info()`, `_logger.debug()`
to log messages during operation execution.
_tracemalloc_started (bool): Internal flag indicating whether tracemalloc
was started. Set to True automatically if logger is enabled for INFO
level or higher.
Notes:
- Memory tracking is only enabled if the logger is configured to handle
INFO level messages or higher. If you want to disable memory tracking,
set the logger level to WARNING or higher.
- Tracemalloc is automatically stopped when the operation object is
deleted (in `__del__`), even if an exception occurs.
- On Windows, pympler may not be available, so object memory tracking
will fall back gracefully. psutil is available on all platforms.
Examples:
Enabling memory tracking for an operation:
>>> import logging
>>> from phenotypic.detect import OtsuDetector
>>> # Set up logging to see memory usage
>>> logging.basicConfig(level=logging.INFO)
>>> # Create detector instance
>>> detector = OtsuDetector()
>>> # Apply operation - memory usage is logged automatically
>>> result = detector.apply(image)
# Console output shows:
# INFO: Memory usage after <step>: XX.XX MB (objects), YY.YY MB (process)
Accessing memory information programmatically:
>>> import logging
>>> from phenotypic.enhance import GaussianBlur
>>> # Create custom logger to capture memory messages
>>> logger = logging.getLogger('phenotypic.enhance.GaussianBlur')
>>> logger.setLevel(logging.INFO)
>>> handler = logging.StreamHandler()
>>> handler.setLevel(logging.INFO)
>>> logger.addHandler(handler)
>>> # Use operation
>>> blur = GaussianBlur(sigma=2)
>>> enhanced = blur.apply(image)
# Memory tracking happens automatically during operation
Custom operation with parameter matching for parallel execution:
>>> from phenotypic.abc_ import ImageOperation
>>> from phenotypic import Image
>>> class CustomThreshold(ImageOperation):
... def __init__(self, threshold_value: int):
... super().__init__()
... self.threshold_value = threshold_value
...
... @staticmethod
... def _operate(image: Image, threshold_value: int = 128) -> Image:
... # Apply threshold algorithm
... image.detect_mat[:] = image.detect_mat[:] > threshold_value
... return image
>>> # When operation is applied via pipeline:
>>> operation = CustomThreshold(threshold_value=100)
# The operation object is serialized with all attributes
# for parallel execution in worker processes
"""
def __init__(self, *args, **kwargs) -> None:
self._logger = logging.getLogger(
f"{self.__class__.__module__}.{self.__class__.__name__}"
)
self._tracemalloc_started = False
# Start tracemalloc automatically if logger is enabled for INFO level
if self._logger.isEnabledFor(logging.INFO):
tracemalloc.start()
self._tracemalloc_started = True
self._logger.debug("Tracemalloc started for memory logging")
def _log_memory_usage(
self,
step: str,
include_process: bool = False,
include_tracemalloc: bool = False,
) -> None:
"""Log memory usage if logger is in INFO mode."""
if self._logger.isEnabledFor(logging.INFO):
log_msg_parts = [f"Memory usage: {step}:"]
# Object memory using pympler
if PYMPLER_AVAILABLE:
try:
all_objects = muppy.get_objects()
mem_summary = summary.summarize(all_objects)
object_memory = sum(
mem[2] for mem in mem_summary
) # mem[2] is total size
log_msg_parts.append(
f"{object_memory / 1024 / 1024:.2f} MB (objects)"
)
except Exception as e:
self._logger.debug(f"Failed to get object memory: {e}")
else:
log_msg_parts.append("pympler not available")
# Process memory using psutil
if include_process and PSUTIL_AVAILABLE:
try:
process = psutil.Process()
process_memory = process.memory_info().rss
log_msg_parts.append(
f"{process_memory / 1024 / 1024:.2f} MB (process)"
)
except Exception as e:
self._logger.debug(f"Failed to get process memory: {e}")
# Tracemalloc snapshot
if include_tracemalloc:
try:
current, peak = tracemalloc.get_traced_memory()
log_msg_parts.append(
f"{current / 1024 / 1024:.2f} MB current, {peak / 1024 / 1024:.2f} MB peak (tracemalloc)"
)
except Exception as e:
self._logger.debug(f"Failed to get tracemalloc memory: {e}")
log_msg = ", ".join(log_msg_parts)
self._logger.info(log_msg)
[docs]
def __del__(self):
"""Automatically stop tracemalloc when the object is deleted."""
if hasattr(self, "_tracemalloc_started") and self._tracemalloc_started:
try:
tracemalloc.stop()
# Only log if we can determine logging is still available
if hasattr(self, "_logger") and hasattr(self._logger, "isEnabledFor"):
self._logger.debug("Tracemalloc stopped automatically")
except Exception:
# Ignore errors during cleanup
pass
