altametris.sara.core ==================== .. py:module:: altametris.sara.core .. autoapi-nested-parse:: Core abstractions partagées par toutes les bibliothèques ML. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/altametris/sara/core/base_callback/index /autoapi/altametris/sara/core/base_detector/index /autoapi/altametris/sara/core/base_exporter/index /autoapi/altametris/sara/core/base_model/index /autoapi/altametris/sara/core/base_trainer/index /autoapi/altametris/sara/core/exceptions/index Exceptions ---------- .. autoapisummary:: altametris.sara.core.AltametrisException altametris.sara.core.ConfigurationError altametris.sara.core.ExportError altametris.sara.core.InferenceError altametris.sara.core.ModelError altametris.sara.core.TrainingError Classes ------- .. autoapisummary:: altametris.sara.core.BaseCallback altametris.sara.core.CallbackManager altametris.sara.core.BaseDetector altametris.sara.core.BaseExporter altametris.sara.core.BaseModel altametris.sara.core.BaseTrainer Package Contents ---------------- .. py:class:: BaseCallback Bases: :py:obj:`abc.ABC` Base class for training callbacks. Callbacks provide hooks into the training lifecycle, allowing custom behavior at key points without modifying training code. All hook methods are optional - override only the ones you need. Hook execution order:: on_train_start() -> for each epoch: -> on_epoch_start() -> for each batch: -> on_batch_start() -> on_batch_end() -> on_validation_start() -> on_validation_end() -> on_epoch_end() on_train_end() .. rubric:: Example >>> class LoggingCallback(BaseCallback): ... def on_epoch_end(self, epoch, metrics): ... print(f"Epoch {epoch}: {metrics}") ... >>> callback = LoggingCallback() >>> callback.on_epoch_end(epoch=1, metrics={"loss": 0.5}) .. py:method:: on_train_start(**kwargs: Any) -> None Called at the start of training. :param \*\*kwargs: Additional context (model, config, etc.) .. rubric:: Example >>> def on_train_start(self, model, epochs, **kwargs): ... print(f"Starting training for {epochs} epochs") .. py:method:: on_train_end(**kwargs: Any) -> None Called at the end of training. :param \*\*kwargs: Additional context (final metrics, model path, etc.) .. rubric:: Example >>> def on_train_end(self, metrics, **kwargs): ... print(f"Training complete: {metrics}") .. py:method:: on_epoch_start(epoch: int, **kwargs: Any) -> None Called at the start of each epoch. :param epoch: Current epoch number (0-indexed) :param \*\*kwargs: Additional context .. rubric:: Example >>> def on_epoch_start(self, epoch, **kwargs): ... print(f"Starting epoch {epoch}") .. py:method:: on_epoch_end(epoch: int, metrics: dict[str, Any], **kwargs: Any) -> None Called at the end of each epoch. :param epoch: Current epoch number (0-indexed) :param metrics: Metrics computed during epoch (loss, mAP, etc.) :param \*\*kwargs: Additional context .. rubric:: Example >>> def on_epoch_end(self, epoch, metrics, **kwargs): ... if metrics["loss"] < 0.1: ... print("Loss threshold reached!") .. py:method:: on_batch_start(batch: int, **kwargs: Any) -> None Called at the start of each batch. :param batch: Current batch number (0-indexed within epoch) :param \*\*kwargs: Additional context (batch data, etc.) .. rubric:: Example >>> def on_batch_start(self, batch, **kwargs): ... if batch % 100 == 0: ... print(f"Processing batch {batch}") .. py:method:: on_batch_end(batch: int, metrics: dict[str, Any], **kwargs: Any) -> None Called at the end of each batch. :param batch: Current batch number (0-indexed within epoch) :param metrics: Batch metrics (batch_loss, etc.) :param \*\*kwargs: Additional context .. rubric:: Example >>> def on_batch_end(self, batch, metrics, **kwargs): ... if metrics["batch_loss"] > 10.0: ... raise TrainingError("Loss explosion detected") .. py:method:: on_validation_start(**kwargs: Any) -> None Called at the start of validation. :param \*\*kwargs: Additional context .. rubric:: Example >>> def on_validation_start(self, **kwargs): ... print("Starting validation...") .. py:method:: on_validation_end(metrics: dict[str, Any], **kwargs: Any) -> None Called at the end of validation. :param metrics: Validation metrics (val_loss, mAP, etc.) :param \*\*kwargs: Additional context .. rubric:: Example >>> def on_validation_end(self, metrics, **kwargs): ... print(f"Validation mAP: {metrics.get('mAP')}") .. py:class:: CallbackManager(callbacks: Optional[list[BaseCallback]] = None) Manages multiple callbacks and orchestrates their execution. The manager ensures all callbacks receive lifecycle events in order, handles exceptions gracefully, and provides logging for debugging. :param callbacks: List of callback instances to manage .. rubric:: Example >>> callback1 = LoggingCallback() >>> callback2 = MetricsCallback() >>> manager = CallbackManager([callback1, callback2]) >>> manager.on_epoch_end(epoch=1, metrics={"loss": 0.5}) .. py:attribute:: callbacks :value: [] .. py:method:: _validate_callbacks() -> None Validate that all callbacks are BaseCallback instances. :raises TrainingError: If any callback is not a BaseCallback instance .. py:method:: add_callback(callback: BaseCallback) -> None Add a callback to the manager. :param callback: Callback instance to add :raises TrainingError: If callback is not a BaseCallback instance .. rubric:: Example >>> manager = CallbackManager() >>> manager.add_callback(LoggingCallback()) .. py:method:: remove_callback(callback: BaseCallback) -> None Remove a callback from the manager. :param callback: Callback instance to remove .. rubric:: Example >>> manager.remove_callback(callback1) .. py:method:: _execute_callback_hook(hook_name: str, *args: Any, **kwargs: Any) -> None Execute a specific hook on all callbacks. :param hook_name: Name of the hook method to call :param \*args: Positional arguments for the hook :param \*\*kwargs: Keyword arguments for the hook :raises TrainingError: If any callback raises an error (re-raised with context) .. py:method:: on_train_start(**kwargs: Any) -> None Execute on_train_start hook on all callbacks. .. py:method:: on_train_end(**kwargs: Any) -> None Execute on_train_end hook on all callbacks. .. py:method:: on_epoch_start(epoch: int, **kwargs: Any) -> None Execute on_epoch_start hook on all callbacks. .. py:method:: on_epoch_end(epoch: int, metrics: dict[str, Any], **kwargs: Any) -> None Execute on_epoch_end hook on all callbacks. .. py:method:: on_batch_start(batch: int, **kwargs: Any) -> None Execute on_batch_start hook on all callbacks. .. py:method:: on_batch_end(batch: int, metrics: dict[str, Any], **kwargs: Any) -> None Execute on_batch_end hook on all callbacks. .. py:method:: on_validation_start(**kwargs: Any) -> None Execute on_validation_start hook on all callbacks. .. py:method:: on_validation_end(metrics: dict[str, Any], **kwargs: Any) -> None Execute on_validation_end hook on all callbacks. .. py:method:: __len__() -> int Return number of registered callbacks. .. py:method:: __repr__() -> str Return string representation of callback manager. .. py:class:: BaseDetector(model_path: Union[str, pathlib.Path], device: str = 'auto', warmup: bool = False, **kwargs: Any) Bases: :py:obj:`abc.ABC` Abstract base class for model inference/detection. Provides common inference infrastructure: - Model loading and initialization - Device management - Warmup capability - Prediction interface :param model_path: Path to model weights :param device: Device for inference ("cpu", "cuda", "mps", "auto") :param warmup: Whether to run warmup inference on initialization .. rubric:: Example >>> detector = MyDetector(model_path="weights/best.pt", device="cuda") >>> results = detector.predict(source="image.jpg") .. py:attribute:: model_path .. py:attribute:: _device .. py:attribute:: model :value: None .. py:attribute:: _is_initialized :value: False .. py:method:: _resolve_device(device: str) -> torch.device Resolve device string to torch.device. :param device: Device string :returns: Resolved torch.device .. py:property:: device :type: torch.device Get current device. .. py:property:: is_initialized :type: bool Check if detector is initialized. .. py:method:: _load_model(model_path: pathlib.Path, **kwargs: Any) -> None :abstractmethod: Load model from path. :param model_path: Path to model file :param \*\*kwargs: Additional loading arguments :raises ModelError: If model cannot be loaded .. note:: Must be implemented by subclasses. Should set self.model. .. py:method:: predict(source: Any, **kwargs: Any) -> Any :abstractmethod: Run inference on source. :param source: Input source (image path, array, video, etc.) :param \*\*kwargs: Inference parameters (conf, iou, etc.) :returns: Prediction results (format depends on detector type) :raises InferenceError: If prediction fails .. note:: Must be implemented by subclasses .. py:method:: warmup(iterations: int = 3) -> None Warmup the model with dummy inference. Useful for GPU models to pre-allocate memory and compile kernels. :param iterations: Number of warmup iterations .. rubric:: Example >>> detector.warmup(iterations=5) .. py:method:: validate_source(source: Any) -> None Validate input source. :param source: Input source to validate :raises InferenceError: If source is invalid .. py:method:: __call__(source: Any, **kwargs: Any) -> Any Callable interface for prediction. :param source: Input source :param \*\*kwargs: Inference parameters :returns: Prediction results .. rubric:: Example >>> detector = MyDetector(model_path="weights/best.pt") >>> results = detector("image.jpg", conf=0.5) .. py:method:: __repr__() -> str String representation of detector. .. py:class:: BaseExporter(model_path: Union[str, pathlib.Path], output_dir: Optional[Union[str, pathlib.Path]] = None) Bases: :py:obj:`abc.ABC` Abstract base class for model export. Provides common export infrastructure: - Multi-format export support - Export validation - Path management - Format-specific configuration :param model_path: Path to source model :param output_dir: Directory for exported models .. rubric:: Example >>> exporter = MyExporter(model_path="weights/best.pt") >>> onnx_path = exporter.export(format="onnx", imgsz=640) >>> exporter.validate_export(onnx_path) .. py:attribute:: SUPPORTED_FORMATS :value: ['onnx', 'torchscript', 'tensorrt', 'coreml', 'tflite'] .. py:attribute:: model_path .. py:attribute:: output_dir .. py:method:: export(format: str = 'onnx', **kwargs: Any) -> pathlib.Path :abstractmethod: Export model to specified format. :param format: Export format (onnx, tensorrt, etc.) :param \*\*kwargs: Format-specific export arguments :returns: Path to exported model :raises ExportError: If export fails .. note:: Must be implemented by subclasses .. py:method:: validate_export(export_path: pathlib.Path, **kwargs: Any) -> bool :abstractmethod: Validate exported model. :param export_path: Path to exported model :param \*\*kwargs: Validation arguments :returns: True if validation passes :raises ExportError: If validation fails .. note:: Must be implemented by subclasses .. py:method:: validate_format(format: str) -> None Validate export format is supported. :param format: Format string to validate :raises ExportError: If format is not supported .. py:method:: get_export_path(format: str, suffix: Optional[str] = None) -> pathlib.Path Generate export file path. :param format: Export format :param suffix: Optional suffix before extension :returns: Path for exported model .. rubric:: Example >>> exporter.get_export_path("onnx") Path("weights/best.onnx") >>> exporter.get_export_path("onnx", suffix="_fp16") Path("weights/best_fp16.onnx") .. py:method:: cleanup_export(export_path: pathlib.Path) -> None Clean up temporary export files. :param export_path: Path to export file to remove .. rubric:: Example >>> exporter.cleanup_export(Path("weights/temp.onnx")) .. py:method:: export_all(formats: Optional[list[str]] = None, **kwargs: Any) -> dict[str, Optional[pathlib.Path]] Export model to multiple formats. :param formats: List of formats to export (default: all supported) :param \*\*kwargs: Export arguments :returns: Dictionary mapping format to export path (None if export failed) .. rubric:: Example >>> exports = exporter.export_all(formats=["onnx", "tensorrt"]) >>> print(exports["onnx"]) .. py:method:: __repr__() -> str String representation of exporter. .. py:class:: BaseModel(config: Optional[dict[str, Any]] = None, device: str = 'auto') Bases: :py:obj:`abc.ABC`, :py:obj:`torch.nn.Module` Abstract base class for all ML models. Provides common functionality for model lifecycle: - Model initialization and configuration - Weight loading and saving - Device management (CPU, CUDA, MPS) - Forward pass abstraction All concrete model implementations must inherit from this class and implement the abstract methods. :param config: Model configuration dictionary :param device: Device to run model on ("cpu", "cuda", "mps", or "auto") .. rubric:: Example >>> class YoloModel(BaseModel): ... def forward(self, x): ... return self.model(x) ... def load_weights(self, path): ... self.model = YOLO(path) ... def save_weights(self, path): ... self.model.save(path) .. py:attribute:: config .. py:attribute:: _device .. py:attribute:: _is_initialized :value: False .. py:method:: _resolve_device(device: str) -> torch.device Resolve device string to torch.device. :param device: Device string ("cpu", "cuda", "mps", "auto") :returns: Resolved torch.device :raises ModelError: If device is invalid or not available .. py:property:: device :type: torch.device Get current device. .. py:method:: to_device(device: Union[str, torch.device]) -> BaseModel Move model to specified device. :param device: Target device :returns: Self for chaining .. rubric:: Example >>> model.to_device("cuda") .. py:method:: validate_config(config: dict[str, Any]) -> None Validate model configuration. :param config: Configuration to validate :raises ConfigurationError: If configuration is invalid .. rubric:: Example >>> model.validate_config({"input_size": 640}) .. py:method:: forward(x: torch.Tensor, **kwargs: Any) -> Any :abstractmethod: Forward pass through the model. :param x: Input tensor :param \*\*kwargs: Additional arguments :returns: Model output .. note:: Must be implemented by subclasses .. py:method:: load_weights(path: Union[str, pathlib.Path], **kwargs: Any) -> None :abstractmethod: Load model weights from file. :param path: Path to weights file :param \*\*kwargs: Additional loading arguments :raises ModelError: If weights cannot be loaded .. note:: Must be implemented by subclasses .. py:method:: save_weights(path: Union[str, pathlib.Path], **kwargs: Any) -> None :abstractmethod: Save model weights to file. :param path: Path to save weights :param \*\*kwargs: Additional saving arguments :raises ModelError: If weights cannot be saved .. note:: Must be implemented by subclasses .. py:method:: get_num_parameters() -> int Get total number of model parameters. :returns: Total parameter count .. rubric:: Example >>> num_params = model.get_num_parameters() >>> print(f"Model has {num_params:,} parameters") .. py:method:: get_trainable_parameters() -> int Get number of trainable parameters. :returns: Trainable parameter count .. py:method:: freeze() -> None Freeze all model parameters (set requires_grad=False). .. rubric:: Example >>> model.freeze() >>> # Model parameters won't be updated during training .. py:method:: unfreeze() -> None Unfreeze all model parameters (set requires_grad=True). .. rubric:: Example >>> model.unfreeze() >>> # Model parameters will be updated during training .. py:method:: summary() -> dict[str, Any] Get model summary with key statistics. :returns: Dictionary with model info .. rubric:: Example >>> summary = model.summary() >>> print(f"Parameters: {summary['total_parameters']}") .. py:method:: __repr__() -> str String representation of model. .. py:class:: BaseTrainer(model: altametris.sara.core.base_model.BaseModel, device: str = 'auto', callbacks: Optional[list[altametris.sara.core.base_callback.BaseCallback]] = None) Bases: :py:obj:`abc.ABC` Abstract base class for model training. Provides common training infrastructure: - Callback management for training lifecycle events - Device management - Training state tracking - Configuration validation :param model: Model to train :param device: Device to train on ("cpu", "cuda", "mps", "auto") :param callbacks: List of callbacks for training events .. rubric:: Example >>> model = MyModel() >>> trainer = MyTrainer(model=model, device="cuda") >>> trainer.train(dataset_config="data.yaml", epochs=100) .. py:attribute:: model .. py:attribute:: device :value: 'auto' .. py:attribute:: callback_manager .. py:attribute:: _is_training :value: False .. py:attribute:: _current_epoch :value: 0 .. py:method:: train(dataset_config: Any, epochs: int = 100, **kwargs: Any) -> dict[str, Any] :abstractmethod: Train the model. :param dataset_config: Dataset configuration (path, dict, etc.) :param epochs: Number of training epochs :param \*\*kwargs: Additional training arguments :returns: Training results and metrics :raises TrainingError: If training fails .. note:: Must be implemented by subclasses .. py:method:: validate(dataset_config: Any, **kwargs: Any) -> dict[str, Any] :abstractmethod: Validate the model. :param dataset_config: Validation dataset configuration :param \*\*kwargs: Additional validation arguments :returns: Validation metrics :raises TrainingError: If validation fails .. note:: Must be implemented by subclasses .. py:method:: add_callback(callback: altametris.sara.core.base_callback.BaseCallback) -> None Add a callback to the trainer. :param callback: Callback instance .. rubric:: Example >>> trainer.add_callback(MyCallback()) .. py:method:: remove_callback(callback: altametris.sara.core.base_callback.BaseCallback) -> None Remove a callback from the trainer. :param callback: Callback instance to remove .. py:property:: is_training :type: bool Check if currently training. .. py:property:: current_epoch :type: int Get current epoch number. .. py:method:: _validate_dataset_config(config: Any) -> None Validate dataset configuration. :param config: Dataset configuration to validate :raises TrainingError: If configuration is invalid .. py:method:: _validate_training_args(epochs: int, **kwargs: Any) -> None Validate training arguments. :param epochs: Number of epochs :param \*\*kwargs: Additional arguments :raises TrainingError: If arguments are invalid .. py:method:: save_checkpoint(path: pathlib.Path, epoch: int, metrics: dict[str, Any]) -> None Save training checkpoint. :param path: Path to save checkpoint :param epoch: Current epoch :param metrics: Current metrics .. rubric:: Example >>> trainer.save_checkpoint( ... path=Path("checkpoints/epoch_10.pt"), ... epoch=10, ... metrics={"loss": 0.5} ... ) .. py:method:: load_checkpoint(path: pathlib.Path) -> dict[str, Any] Load training checkpoint. :param path: Path to checkpoint file :returns: Checkpoint data :raises TrainingError: If checkpoint cannot be loaded .. py:method:: __repr__() -> str String representation of trainer. .. py:exception:: AltametrisException(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`Exception` Base exception for all Altametris libraries. All custom exceptions inherit from this base class to allow catching all Altametris-specific errors with a single except clause. :param message: Error message describing what went wrong :param details: Optional dictionary with additional error context .. rubric:: Example >>> try: ... raise AltametrisException("Something went wrong", {"code": 500}) ... except AltametrisException as e: ... print(e) ... print(e.details) .. py:attribute:: message .. py:attribute:: details .. py:method:: __str__() -> str Return string representation of exception. .. py:method:: __repr__() -> str Return repr representation of exception. .. py:exception:: ConfigurationError(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`AltametrisException` Exception raised for configuration-related errors. Scenarios: - Invalid configuration parameter - Missing required configuration key - Configuration validation failed - Incompatible configuration values - Invalid YAML/JSON format .. rubric:: Example >>> raise ConfigurationError( ... "Invalid batch_size", ... {"parameter": "batch_size", "value": -1, "expected": "> 0"} ... ) .. py:exception:: ExportError(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`AltametrisException` Exception raised for export-related errors. Scenarios: - Unsupported export format - Export validation failed - Missing export dependencies (onnx, tensorrt) - Dynamic shapes not supported - Export optimization failed .. rubric:: Example >>> raise ExportError( ... "ONNX export failed", ... {"format": "onnx", "reason": "Dynamic axes not supported"} ... ) .. py:exception:: InferenceError(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`AltametrisException` Exception raised for inference-related errors. Scenarios: - Invalid input format (wrong shape, type) - Device mismatch (model on CUDA, input on CPU) - Prediction failure - Post-processing error - Batch size too large for memory .. rubric:: Example >>> raise InferenceError( ... "Invalid input shape", ... {"expected": (3, 640, 640), "got": (640, 640)} ... ) .. py:exception:: ModelError(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`AltametrisException` Exception raised for model-related errors. Scenarios: - Failed to load model weights - Invalid model architecture - Corrupted checkpoint file - Incompatible model version - Missing required model files .. rubric:: Example >>> raise ModelError( ... "Failed to load weights", ... {"path": "model.pt", "reason": "File not found"} ... ) .. py:exception:: TrainingError(message: str, details: Optional[dict[str, Any]] = None) Bases: :py:obj:`AltametrisException` Exception raised for training-related errors. Scenarios: - Dataset not found or invalid format - Training divergence (NaN loss) - Out of memory during training - Invalid hyperparameters - Checkpoint save failure .. rubric:: Example >>> raise TrainingError( ... "Training diverged", ... {"epoch": 10, "loss": float('nan'), "lr": 0.001} ... )