Skip to content

API Reference: CoreModel

High-level unified interface for CoreCV models. Wraps model initialization, training loops, inference pipelines, and model export formats under a single facade.


CoreModel

corecv.api.model.CoreModel(model, task=None, input_size=_DEFAULT_INPUT_SIZE, device=None, num_classes=None, pretrained=True, neck=None, head=None, **kwargs)

High-level unified API facade for CoreCV models.

Wraps any CoreCV model (classification, segmentation, or detection) and exposes a single entry point for training, inference, and export through delegation to specialised engines.

Engines are created lazily — no heavyweight initialisation happens until the corresponding method is called.

Parameters:

Name Type Description Default
model Module | str | Path | dict[str, Any]

One of:

  • nn.Module — a pre-built CoreCV model.
  • str / Path — path to a .pt / .pth checkpoint, a .yaml / .yml configuration file, or a plain registered backbone name (e.g. "resnet18").
  • dict — raw configuration dictionary containing at minimum model_name.
required
task Literal['classification', 'segmentation', 'detection'] | None

Task type. One of "classification", "segmentation", or "detection".

None
input_size tuple[int, int]

Input image dimensions (height, width). Default (224, 224).

_DEFAULT_INPUT_SIZE
device device | None

Target :class:torch.device. If None, auto-detects CUDA when available.

None
num_classes int | None

Number of output classes. If None, inferred from the model (via model.head.num_classes) when possible.

None
Example

import torch from corecv.api import CoreModel from corecv.models import CoreObjectDetector

detector = CoreObjectDetector(...) model = CoreModel(detector, task="detection", input_size=(640, 640))

Fluent configuration

(model ... .set_loss_fn(torch.nn.CrossEntropyLoss()) ... .set_train_dataloader(train_loader) ... .set_val_dataloader(val_loader))

Train

history = model.train(epochs=50, lr=0.001, batch_size=16)

Predict

results = model.predict("test.jpg", conf_threshold=0.5)

Export

paths = model.export(format="onnx", target_hardware="edge")

Initialise the CoreModel facade.

Parameters:

Name Type Description Default
model Module | str | Path | dict[str, Any]

A CoreCV model (nn.Module), a path to a .pt / .pth checkpoint, a path to a .yaml / .yml configuration file, a plain registered backbone name (e.g. "resnet18"), or a raw configuration dict containing model and training specifications.

required
task Literal['classification', 'segmentation', 'detection'] | None

Task type discriminant. If None, inferred from the configuration dictionary / YAML file, defaulting to "classification".

None
input_size tuple[int, int]

Input (height, width).

_DEFAULT_INPUT_SIZE
device device | None

Target device (auto-detected if None).

None
num_classes int | None

Number of output classes (inferred if None).

None
pretrained bool

Whether to load pretrained backbone weights.

True
neck str | None

Registered neck name (e.g. "fpn", "panet").

None
head str | None

Registered head name (e.g. "decoupled_anchor_free").

None
**kwargs Any

Additional configuration entries forwarded to component constructors and stored for training execution.

{}
Source code in src/corecv/api/model.py
def __init__(  # noqa: PLR0913, PLR0912, PLR0915
    self,
    model: nn.Module | str | Path | dict[str, Any],
    task: Literal["classification", "segmentation", "detection"] | None = None,
    input_size: tuple[int, int] = _DEFAULT_INPUT_SIZE,
    device: torch.device | None = None,
    num_classes: int | None = None,
    pretrained: bool = True,
    neck: str | None = None,
    head: str | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Initialise the CoreModel facade.

    Args:
        model: A CoreCV model (``nn.Module``), a path to a ``.pt`` /
            ``.pth`` checkpoint, a path to a ``.yaml`` / ``.yml``
            configuration file, a plain registered backbone name
            (e.g. ``"resnet18"``), or a raw configuration ``dict``
            containing model and training specifications.
        task: Task type discriminant. If ``None``, inferred from the
            configuration dictionary / YAML file, defaulting to
            ``"classification"``.
        input_size: Input ``(height, width)``.
        device: Target device (auto-detected if ``None``).
        num_classes: Number of output classes (inferred if ``None``).
        pretrained: Whether to load pretrained backbone weights.
        neck: Registered neck name (e.g. ``"fpn"``, ``"panet"``).
        head: Registered head name (e.g. ``"decoupled_anchor_free"``).
        **kwargs: Additional configuration entries forwarded to component
            constructors and stored for training execution.
    """
    self._config_dict: dict[str, Any] = {}

    # --- Polymorphic model loading --------------------------------
    if isinstance(model, dict):
        t = task or model.get("task", "classification")
        merged = {**model, "task": t, "pretrained": pretrained}
        if num_classes is not None:
            merged["num_classes"] = num_classes
        if neck is not None:
            merged["neck_type"] = neck
        if head is not None:
            merged["head_type"] = head
        merged.update(kwargs)
        self._config_dict = dict(merged)
        self._model = self._build_model_from_config(merged)
        task = t
    elif isinstance(model, (str, Path)):
        path = Path(model)
        ext = path.suffix.lower()
        if ext in (".pt", ".pth"):
            checkpoint_config = self._load_from_checkpoint(path)
            self._config_dict = dict(checkpoint_config)
            num_classes = (
                num_classes
                if num_classes is not None
                else checkpoint_config.get("num_classes")
            )
            t = task or checkpoint_config.get("task", "classification")
            task = t
        elif ext in (".yaml", ".yml"):
            raw_cfg = self._load_from_config(path)
            t = task or raw_cfg.get("task", "classification")
            self._config_dict = dict(raw_cfg)
            if num_classes is not None:
                self._config_dict["num_classes"] = num_classes
            self._config_dict.update(kwargs)
            task = t
        elif ext == "":
            t = task or "classification"
            config: dict[str, Any] = {
                "model_name": str(model),
                "task": t,
                "num_classes": num_classes or 1000,
                "pretrained": pretrained,
            }
            if neck is not None:
                config["neck_type"] = neck
            if head is not None:
                config["head_type"] = head
            config.update(kwargs)
            self._config_dict = dict(config)
            self._model = self._build_model_from_config(config)
            task = t
        else:
            raise ValueError(
                _ERR_MODEL_EXTENSION.format(ext)
            )
    elif isinstance(model, nn.Module):
        self._model = model
        task = task or "classification"
    else:
        raise TypeError(
            _ERR_MODEL_TYPE.format(type(model).__name__)
        )

    if task not in _SUPPORTED_TASKS:
        raise ValueError(
            _ERR_UNSUPPORTED_TASK.format(task, sorted(_SUPPORTED_TASKS))
        )

    # --- Core attributes -------------------------------------------
    self._task: str = task
    self._input_size: tuple[int, int] = (
        int(input_size[0]), int(input_size[1])
    )
    self._device: torch.device = (
        device
        if device is not None
        else torch.device("cuda" if torch.cuda.is_available() else "cpu")
    )
    self._num_classes: int | None = (
        num_classes if num_classes is not None else self._infer_num_classes()
    )

    # --- Optional components (must be set before training) ---------
    self._loss_fn: object | None = None
    self._train_loader: DataLoader | None = None
    self._val_loader: DataLoader | None = None

    # --- Lazy engines ----------------------------------------------
    self._trainer: CoreTrainer | None = None
    self._predictor: CorePredictor | None = None
    self._exporter: CoreExporter | None = None

device property

Return the target :class:torch.device.

input_size property

Return the input (height, width) used for preprocessing.

model property

Return the wrapped CoreCV model.

num_classes property

Return the number of output classes, or None if unknown.

predictor property

Return the internal :class:CorePredictor instance.

None until :meth:predict is called.

task property

Return the task type.

One of "classification", "segmentation", or "detection".

trainer property

Return the internal :class:CoreTrainer instance.

None until :meth:train is called.

export(format='onnx', target_hardware='server', opset=_DEFAULT_OPSET, optimize=True, output_path=None, input_shape=None, dynamic_axes=None, weights=None)

Export the model to ONNX and/or ExecuTorch format.

Delegates to :class:CoreExporter which internally uses :class:TargetRewriter (for edge-hardware graph rewrites) and :class:MetaProber (for zero-VRAM shape validation).

The export pipeline is:

  1. Rewrite — When target_hardware='edge', applies activation replacements (GELU -> ReLU, SiLU -> Hardswish) and collapses redundant layout permutations.
  2. Validate — Runs shape propagation on device='meta' and audits the graph for dynamic operations.
  3. Export — Serialises to .onnx and/or .pte.

Parameters:

Name Type Description Default
format str

Export target. One of "onnx", "executorch", "both".

'onnx'
target_hardware str

Hardware profile. "edge" applies activation rewrites; "server" skips them.

'server'
opset int

ONNX opset version (17 or 18).

_DEFAULT_OPSET
optimize bool

When True, enables TargetRewriter graph optimisations and XNNPACK delegate (for ExecuTorch).

True
output_path str | None

Explicit output file path. For "both" format, this is used as a prefix (e.g. "model" produces "model.onnx" and "model.pte"). If None, a timestamped path is auto-generated.

None
input_shape tuple[int, ...] | None

Input tensor shape (B, C, H, W). Defaults to (1, 3, H, W) using self.input_size.

None
dynamic_axes dict[str, dict[int, str]] | None

ONNX-style dynamic axes dictionary. None means static shapes.

None
weights str | Path | None

Optional path to a .pt / .pth checkpoint. If provided, the model weights are loaded from this file before export. The architecture must already be set (e.g. via config or a previous checkpoint load).

None

Returns:

Type Description
dict[str, str]

A dictionary mapping format names to file paths, e.g.

dict[str, str]

{"onnx": "/path/to/model.onnx"}.

Raises:

Type Description
ValueError

If any parameter is invalid.

RuntimeError

If validation or export fails.

Example::

>>> # ONNX for server
>>> paths = model.export(format="onnx", target_hardware="server")
>>> paths["onnx"]
'.../model_20260723_021130.onnx'

>>> # ExecuTorch for edge with rewrites
>>> paths = model.export(
...     format="executorch",
...     target_hardware="edge",
...     opset=18,
... )
>>> paths["executorch"]
'.../model_20260723_021131.pte'

>>> # Both formats
>>> paths = model.export(format="both")
>>> list(paths.keys())
['onnx', 'executorch']

>>> # Load weights before export
>>> paths = model.export(format="onnx", weights="best.pt")
Source code in src/corecv/api/model.py
def export(  # noqa: PLR0913
    self,
    format: str = "onnx",
    target_hardware: str = "server",
    opset: int = _DEFAULT_OPSET,
    optimize: bool = True,
    output_path: str | None = None,
    input_shape: tuple[int, ...] | None = None,
    dynamic_axes: dict[str, dict[int, str]] | None = None,
    weights: str | Path | None = None,
) -> dict[str, str]:
    """Export the model to ONNX and/or ExecuTorch format.

    Delegates to :class:`CoreExporter` which internally uses
    :class:`TargetRewriter` (for edge-hardware graph rewrites)
    and :class:`MetaProber` (for zero-VRAM shape validation).

    The export pipeline is:

    1. **Rewrite** — When ``target_hardware='edge'``, applies
       activation replacements (GELU -> ReLU, SiLU -> Hardswish)
       and collapses redundant layout permutations.
    2. **Validate** — Runs shape propagation on ``device='meta'``
       and audits the graph for dynamic operations.
    3. **Export** — Serialises to ``.onnx`` and/or ``.pte``.

    Args:
        format: Export target.  One of ``"onnx"``, ``"executorch"``,
            ``"both"``.
        target_hardware: Hardware profile.  ``"edge"`` applies
            activation rewrites; ``"server"`` skips them.
        opset: ONNX opset version (``17`` or ``18``).
        optimize: When ``True``, enables TargetRewriter graph
            optimisations and XNNPACK delegate (for ExecuTorch).
        output_path: Explicit output file path.  For ``"both"``
            format, this is used as a prefix (e.g. ``"model"``
            produces ``"model.onnx"`` and ``"model.pte"``).
            If ``None``, a timestamped path is auto-generated.
        input_shape: Input tensor shape ``(B, C, H, W)``.
            Defaults to ``(1, 3, H, W)`` using ``self.input_size``.
        dynamic_axes: ONNX-style dynamic axes dictionary.
            ``None`` means static shapes.
        weights: Optional path to a ``.pt`` / ``.pth`` checkpoint.
            If provided, the model weights are loaded from this file
            before export.  The architecture must already be set
            (e.g. via config or a previous checkpoint load).

    Returns:
        A dictionary mapping format names to file paths, e.g.
        ``{"onnx": "/path/to/model.onnx"}``.

    Raises:
        ValueError: If any parameter is invalid.
        RuntimeError: If validation or export fails.

    Example::

        >>> # ONNX for server
        >>> paths = model.export(format="onnx", target_hardware="server")
        >>> paths["onnx"]
        '.../model_20260723_021130.onnx'

        >>> # ExecuTorch for edge with rewrites
        >>> paths = model.export(
        ...     format="executorch",
        ...     target_hardware="edge",
        ...     opset=18,
        ... )
        >>> paths["executorch"]
        '.../model_20260723_021131.pte'

        >>> # Both formats
        >>> paths = model.export(format="both")
        >>> list(paths.keys())
        ['onnx', 'executorch']

        >>> # Load weights before export
        >>> paths = model.export(format="onnx", weights="best.pt")
    """
    # --- Load weights if provided -----------------------------------
    if weights is not None:
        self._load_weights(weights)

    # --- Validate export config -------------------------------------
    export_cfg = self._resolve_export_config(
        format=format,
        target_hardware=target_hardware,
        opset=opset,
        optimize=optimize,
        output_path=output_path,
        input_shape=input_shape,
        dynamic_axes=dynamic_axes,
    )

    # --- Build CoreExporter -----------------------------------------
    ex: CoreExporter = CoreExporter(
        model=self._model,
        target=export_cfg.format,
        opset_version=export_cfg.opset,
        target_hardware=export_cfg.target_hardware,
        input_shape=export_cfg.input_shape,
        dynamic_axes=export_cfg.dynamic_axes,
        output_dir=str(Path(export_cfg.output_path).parent)
        if export_cfg.output_path
        else tempfile.mkdtemp(prefix="corecv_export_"),
    )

    # --- Run export pipeline ----------------------------------------
    logger.info(
        "Starting export (format=%s, hardware=%s, opset=%d, optimize=%s)",
        export_cfg.format,
        export_cfg.target_hardware,
        export_cfg.opset,
        export_cfg.optimize,
    )

    results: dict[str, str] = ex.run_export()

    # --- Rename if explicit output_path was provided ----------------
    if export_cfg.output_path is not None:
        results = self._rename_export_outputs(results, export_cfg.output_path)

    logger.info("Export completed: %s", results)
    return results

from_pretrained(path, device=None) classmethod

Load a pretrained model from a checkpoint file.

The checkpoint must contain a model_config key (a dictionary describing the architecture) and a model_state_dict key containing the trained weights. The architecture is rebuilt from model_config, the weights are loaded, and a fully-configured :class:CoreModel instance is returned.

Parameters:

Name Type Description Default
path str | Path

Path to a .pt or .pth checkpoint file.

required
device device | None

Target device. If None, auto-detected.

None

Returns:

Name Type Description
A CoreModel

class:CoreModel wrapping the reconstructed model.

Raises:

Type Description
FileNotFoundError

If the checkpoint file does not exist.

KeyError

If the checkpoint is missing model_config.

RuntimeError

If the checkpoint cannot be loaded.

Source code in src/corecv/api/model.py
@classmethod
def from_pretrained(
    cls,
    path: str | Path,
    device: torch.device | None = None,
) -> CoreModel:
    """Load a pretrained model from a checkpoint file.

    The checkpoint must contain a ``model_config`` key (a dictionary
    describing the architecture) and a ``model_state_dict`` key
    containing the trained weights.  The architecture is rebuilt from
    ``model_config``, the weights are loaded, and a fully-configured
    :class:`CoreModel` instance is returned.

    Args:
        path: Path to a ``.pt`` or ``.pth`` checkpoint file.
        device: Target device.  If ``None``, auto-detected.

    Returns:
        A :class:`CoreModel` wrapping the reconstructed model.

    Raises:
        FileNotFoundError: If the checkpoint file does not exist.
        KeyError: If the checkpoint is missing ``model_config``.
        RuntimeError: If the checkpoint cannot be loaded.
    """
    path_obj = Path(path)
    if not path_obj.exists():
        raise FileNotFoundError(
            _ERR_CHECKPOINT_NOT_FOUND.format(path_obj)
        )

    checkpoint: dict[str, Any] = torch.load(
        str(path_obj), map_location="cpu", weights_only=False
    )

    model_config: dict[str, Any] | None = checkpoint.get("model_config")
    if model_config is None:
        raise KeyError(
            _ERR_CHECKPOINT_NO_CONFIG.format(
                path_obj, list(checkpoint.keys())
            )
        )

    # Rebuild architecture from config
    model = cls._build_model_from_config(model_config)

    # Load state dict
    state_dict = checkpoint.get(
        "model_state_dict",
        checkpoint.get("state_dict"),
    )
    if state_dict is not None:
        model.load_state_dict(state_dict, strict=False)
        logger.info(
            "Loaded state_dict into model built from config (strict=False)"
        )

    # Infer metadata
    task = model_config.get("task", "classification")
    input_size = model_config.get(
        "input_size",
        (
            model_config.get("input_height", _DEFAULT_INPUT_SIZE[0]),
            model_config.get("input_width", _DEFAULT_INPUT_SIZE[1]),
        ),
    )
    if isinstance(input_size, (list, tuple)) and len(input_size) == _ERR_INPUT_SIZE_DIM:  # noqa: PLR2004
        input_size = (int(input_size[0]), int(input_size[1]))
    else:
        input_size = _DEFAULT_INPUT_SIZE

    num_classes = model_config.get("num_classes")

    return cls(
        model=model,
        task=task,  # type: ignore[arg-type]
        input_size=input_size,
        device=device,
        num_classes=num_classes,
    )

predict(source, conf_threshold=None, iou_threshold=None, topk=None, half_precision=False, compile_model=False, batch_size=8, weights=None)

Run inference on one or more images.

Delegates to :class:CorePredictor which handles preprocessing, GPU-native post-processing, and batching.

Parameters:

Name Type Description Default
source str | Path | Tensor | list[str | Path | Tensor]

Input source — a single image path, a list of paths, a torch.Tensor, or a list of tensors.

required
conf_threshold float | None

Minimum confidence score for detection predictions. None uses the predictor default (0.25).

None
iou_threshold float | None

IoU threshold for NMS in detection. None uses the predictor default (0.45).

None
topk int | None

Number of top predictions for classification. None uses the predictor default (5).

None
half_precision bool

Enable FP16 inference via torch.autocast. Default False.

False
compile_model bool

Enable torch.compile on the model (requires PyTorch >= 2.0). Default False.

False
batch_size int

Maximum batch size for list / folder inference. Default 8.

8
weights str | Path | None

Optional path to a .pt / .pth checkpoint. If provided, the model weights are loaded from this file before prediction. This is useful for swapping weights without rebuilding the :class:CoreModel.

None

Returns:

Type Description
list[Prediction]

A list of :class:Prediction objects, one per input image.

Example::

>>> # Single image
>>> preds = model.predict("photo.jpg", topk=5)
>>> print(preds[0].classification.class_ids)

>>> # Batch of tensors
>>> batch = torch.randn(4, 3, 224, 224)
>>> preds = model.predict(list(batch))

>>> # Load weights before prediction
>>> preds = model.predict("photo.jpg", weights="best.pt")
Source code in src/corecv/api/model.py
def predict(  # noqa: PLR0913
    self,
    source: (
        str | Path | torch.Tensor | list[str | Path | torch.Tensor]
    ),
    conf_threshold: float | None = None,
    iou_threshold: float | None = None,
    topk: int | None = None,
    half_precision: bool = False,
    compile_model: bool = False,
    batch_size: int = 8,
    weights: str | Path | None = None,
) -> list[Prediction]:
    """Run inference on one or more images.

    Delegates to :class:`CorePredictor` which handles preprocessing,
    GPU-native post-processing, and batching.

    Args:
        source: Input source — a single image path, a list of paths,
            a ``torch.Tensor``, or a list of tensors.
        conf_threshold: Minimum confidence score for detection
            predictions.  ``None`` uses the predictor default (0.25).
        iou_threshold: IoU threshold for NMS in detection.
            ``None`` uses the predictor default (0.45).
        topk: Number of top predictions for classification.
            ``None`` uses the predictor default (5).
        half_precision: Enable FP16 inference via ``torch.autocast``.
            Default ``False``.
        compile_model: Enable ``torch.compile`` on the model (requires
            PyTorch >= 2.0).  Default ``False``.
        batch_size: Maximum batch size for list / folder inference.
            Default ``8``.
        weights: Optional path to a ``.pt`` / ``.pth`` checkpoint.
            If provided, the model weights are loaded from this file
            before prediction.  This is useful for swapping weights
            without rebuilding the :class:`CoreModel`.

    Returns:
        A list of :class:`Prediction` objects, one per input image.

    Example::

        >>> # Single image
        >>> preds = model.predict("photo.jpg", topk=5)
        >>> print(preds[0].classification.class_ids)

        >>> # Batch of tensors
        >>> batch = torch.randn(4, 3, 224, 224)
        >>> preds = model.predict(list(batch))

        >>> # Load weights before prediction
        >>> preds = model.predict("photo.jpg", weights="best.pt")
    """
    # Load weights if provided
    if weights is not None:
        self._load_weights(weights)

    # Use provided thresholds or fall back to defaults
    _conf: float = conf_threshold if conf_threshold is not None else 0.25
    _iou: float = iou_threshold if iou_threshold is not None else 0.45
    _topk: int = topk if topk is not None else 5

    # Build predictor lazily or with updated params
    if self._predictor is None or self._needs_predictor_rebuild(
        _conf, _iou, _topk, half_precision, compile_model, batch_size,
    ):
        self._predictor = self._build_predictor(
            conf_threshold=_conf,
            iou_threshold=_iou,
            topk=_topk,
            half_precision=half_precision,
            compile_model=compile_model,
            batch_size=batch_size,
        )

    # Run prediction
    return self._predictor.predict(source)

set_loss_fn(loss_fn)

Set the loss function for training.

Parameters:

Name Type Description Default
loss_fn object

A callable loss_fn(preds, targets) -> Tensor.

required

Returns:

Type Description
CoreModel

self for chaining.

Source code in src/corecv/api/model.py
def set_loss_fn(self, loss_fn: object) -> CoreModel:
    """Set the loss function for training.

    Args:
        loss_fn: A callable ``loss_fn(preds, targets) -> Tensor``.

    Returns:
        ``self`` for chaining.
    """
    self._loss_fn = loss_fn
    return self

set_train_dataloader(loader)

Set the training :class:DataLoader.

Parameters:

Name Type Description Default
loader DataLoader

Training data loader.

required

Returns:

Type Description
CoreModel

self for chaining.

Source code in src/corecv/api/model.py
def set_train_dataloader(self, loader: DataLoader) -> CoreModel:
    """Set the training :class:`DataLoader`.

    Args:
        loader: Training data loader.

    Returns:
        ``self`` for chaining.
    """
    self._train_loader = loader
    return self

set_val_dataloader(loader)

Set the validation :class:DataLoader.

Parameters:

Name Type Description Default
loader DataLoader

Validation data loader.

required

Returns:

Type Description
CoreModel

self for chaining.

Source code in src/corecv/api/model.py
def set_val_dataloader(self, loader: DataLoader) -> CoreModel:
    """Set the validation :class:`DataLoader`.

    Args:
        loader: Validation data loader.

    Returns:
        ``self`` for chaining.
    """
    self._val_loader = loader
    return self

train(config=None, *, target_hardware='server', **kwargs)

train(
    config: str,
    *,
    target_hardware: str = "server",
    **kwargs: object,
) -> dict[str, list]
train(
    config: dict[str, Any],
    *,
    target_hardware: str = "server",
    **kwargs: object,
) -> dict[str, list]
train(
    config: TrainingConfig,
    *,
    target_hardware: str = "server",
    **kwargs: object,
) -> dict[str, list]
train(
    *, target_hardware: str = "server", **kwargs: object
) -> dict[str, list]

Train the model with the given configuration.

Accepts a polymorphic config argument:

  • str — Path to a .yaml configuration file.
  • dict — Configuration dictionary.
  • :class:TrainingConfig — A validated dataclass instance.
  • None — All parameters are provided via **kwargs.

In all cases, keyword arguments take precedence over values in config (when both are present).

Required pre-conditions (must be set before calling train):

  • A train :class:DataLoader via :meth:set_train_dataloader or passed via config.
  • A loss function via :meth:set_loss_fn.

Parameters:

Name Type Description Default
config str | dict[str, Any] | TrainingConfig | None

Path to a .yaml file, a dict, or a :class:TrainingConfig instance. None means all parameters come from **kwargs.

None
target_hardware str

Hardware profile. "edge" applies activation rewrites (GELU -> ReLU, SiLU -> Hardswish) and LayerNorm collapses before building the optimiser. "server" (default) skips rewrites.

'server'
**kwargs object

Additional or overriding training hyperparameters. See :class:TrainingConfig for supported keys.

{}

Returns:

Type Description
dict[str, list]

A history dictionary with keys "train" and "val",

dict[str, list]

each containing a list of per-epoch metric dictionaries.

Raises:

Type Description
ValueError

If configuration validation fails or required components are missing.

RuntimeError

If the training engine encounters an error.

Example::

>>> # Keyword arguments
>>> history = model.train(epochs=10, lr=0.001, batch_size=64)

>>> # Dictionary
>>> history = model.train({"epochs": 10, "lr": 0.001})

>>> # YAML file
>>> history = model.train("configs/train.yaml")

>>> # Dataclass
>>> cfg = TrainingConfig(epochs=10, lr=0.001)
>>> history = model.train(cfg)

>>> # Mixed (kwargs override dict)
>>> history = model.train({"epochs": 10}, batch_size=128)
Source code in src/corecv/api/model.py
def train(
    self,
    config: str | dict[str, Any] | TrainingConfig | None = None,
    *,
    target_hardware: str = "server",
    **kwargs: object,
) -> dict[str, list]:
    """Train the model with the given configuration.

    Accepts a polymorphic ``config`` argument:

    * ``str`` — Path to a ``.yaml`` configuration file.
    * ``dict`` — Configuration dictionary.
    * :class:`TrainingConfig` — A validated dataclass instance.
    * ``None`` — All parameters are provided via ``**kwargs``.

    In all cases, keyword arguments take precedence over values in
    ``config`` (when both are present).

    **Required pre-conditions** (must be set before calling ``train``):

    * A train :class:`DataLoader` via :meth:`set_train_dataloader`
      or passed via ``config``.
    * A loss function via :meth:`set_loss_fn`.

    Args:
        config: Path to a ``.yaml`` file, a ``dict``, or a
            :class:`TrainingConfig` instance.  ``None`` means all
            parameters come from ``**kwargs``.
        target_hardware: Hardware profile.  ``"edge"`` applies
            activation rewrites (GELU -> ReLU, SiLU -> Hardswish)
            and LayerNorm collapses **before** building the
            optimiser.  ``"server"`` (default) skips rewrites.
        **kwargs: Additional or overriding training hyperparameters.
            See :class:`TrainingConfig` for supported keys.

    Returns:
        A history dictionary with keys ``"train"`` and ``"val"``,
        each containing a list of per-epoch metric dictionaries.

    Raises:
        ValueError: If configuration validation fails or required
            components are missing.
        RuntimeError: If the training engine encounters an error.

    Example::

        >>> # Keyword arguments
        >>> history = model.train(epochs=10, lr=0.001, batch_size=64)

        >>> # Dictionary
        >>> history = model.train({"epochs": 10, "lr": 0.001})

        >>> # YAML file
        >>> history = model.train("configs/train.yaml")

        >>> # Dataclass
        >>> cfg = TrainingConfig(epochs=10, lr=0.001)
        >>> history = model.train(cfg)

        >>> # Mixed (kwargs override dict)
        >>> history = model.train({"epochs": 10}, batch_size=128)
    """
    # --- Merge full configuration for auto-building -----------------
    merged_config: dict[str, Any] = dict(self._config_dict)
    if isinstance(config, str):
        import yaml  # noqa: PLC0415
        p_obj = Path(config)
        if p_obj.exists():
            with p_obj.open("r", encoding="utf-8") as f:
                raw_y = yaml.safe_load(f)
            if isinstance(raw_y, dict):
                merged_config.update(raw_y)
    elif isinstance(config, dict):
        merged_config.update(config)
    elif isinstance(config, TrainingConfig):
        merged_config.update(
            {f.name: getattr(config, f.name) for f in dataclasses.fields(config)}
        )
    merged_config.update(kwargs)

    # --- Auto-build missing loss_fn and dataloaders ----------------
    self._auto_build_loss_fn()
    self._auto_build_dataloaders(merged_config)

    # --- Resolve and validate training config -----------------------
    train_cfg: TrainingConfig = self._resolve_train_config(
        config, target_hardware=target_hardware, **kwargs
    )

    # --- Check required components ----------------------------------
    if self._train_loader is None:
        msg = (
            _ERR_MISSING_DATALOADER
            + " Alternatively, provide a 'data' path in config or train(data='...')."
        )
        raise ValueError(msg)
    if self._loss_fn is None:
        raise ValueError(_ERR_MISSING_LOSS_FN)

    # --- Resolve device ---------------------------------------------
    device: torch.device = (
        torch.device(train_cfg.device)
        if train_cfg.device is not None
        else self._device
    )

    # --- Rewrite graph for edge hardware (if requested) ------------
    if train_cfg.target_hardware == "edge":
        logger.info(
            "Applying edge-hardware graph rewrites (GELU -> ReLU,"
            " SiLU -> Hardswish, LayerNorm collapses)"
        )
        self._model = TargetRewriter().rewrite_for_edge(self._model)

    # --- Build optimiser and scheduler ------------------------------
    optimiser: torch.optim.Optimizer = self._build_optimizer(train_cfg)
    scheduler: object | None = self._build_scheduler(train_cfg, optimiser)

    # --- Create CoreTrainer -----------------------------------------
    self._trainer = CoreTrainer(
        model=self._model,
        optimizer=optimiser,
        loss_fn=self._loss_fn,
        train_dataloader=self._train_loader,
        val_dataloader=self._val_loader,
        device=device,
        gradient_accumulation_steps=train_cfg.grad_accum,
        max_grad_norm=train_cfg.clip_grad,
        use_amp=train_cfg.amp,
        ema_decay=train_cfg.ema_decay if train_cfg.ema else None,
        scheduler=scheduler,
        val_metrics=self._auto_build_val_metrics(device),
        output_dir=train_cfg.output_dir,
    )

    # --- Run training loop ------------------------------------------
    logger.info(
        "Starting training for %d epochs (task=%s, device=%s, lr=%.6f)",
        train_cfg.epochs,
        self._task,
        device.type,
        train_cfg.lr,
    )
    history: dict[str, list] = self._trainer.fit(
        num_epochs=train_cfg.epochs,
    )
    logger.info("Training completed.")
    return history

TrainingConfig

corecv.api.model.TrainingConfig(*, epochs=_DEFAULT_EPOCHS, lr=_DEFAULT_LR, batch_size=_DEFAULT_BATCH_SIZE, optimizer=_DEFAULT_OPTIMIZER, scheduler=None, amp=True, grad_accum=1, clip_grad=1.0, ema=True, ema_decay=0.9999, device=None, output_dir='./checkpoints', target_hardware='server') dataclass

Validated training hyperparameter configuration.

All fields are validated in __post_init__.

Attributes:

Name Type Description
epochs int

Number of training epochs. Must be >= 1.

lr float

Learning rate. Must be > 0.

batch_size int

Batch size per device. Must be >= 1.

optimizer str

Optimizer name. One of "adamw", "adam", "sgd".

scheduler str | None

Scheduler name. One of "cosine", "step", "none", or None.

amp bool

Enable automatic mixed precision. Default True.

grad_accum int

Gradient accumulation steps. Must be >= 1.

clip_grad float | None

Max gradient norm for clipping. None disables clipping. Default 1.0.

ema bool

Enable exponential moving average. Default True.

ema_decay float

EMA decay factor. Must be in (0, 1). Default 0.9999.

device str | None

Target device string (e.g. "cuda", "cpu"). If None, auto-detected.

output_dir str

Directory for checkpoints and logs.

target_hardware str

Hardware profile. "edge" applies activation rewrites (GELU -> ReLU, SiLU -> Hardswish) and LayerNorm collapses before the optimiser is built; "server" (default) skips rewrites.

__post_init__()

Validate training configuration fields.

Source code in src/corecv/api/model.py
def __post_init__(self) -> None:
    """Validate training configuration fields."""
    hw = self.target_hardware.lower()
    object.__setattr__(self, "target_hardware", hw)
    if hw not in _VALID_HARDWARE_TARGETS:
        raise ValueError(
            _ERR_UNKNOWN_HARDWARE.format(self.target_hardware, sorted(_VALID_HARDWARE_TARGETS))
        )
    if self.epochs < 1:
        raise ValueError(_ERR_EPOCHS_RANGE.format(self.epochs))
    if self.lr <= 0.0:
        raise ValueError(_ERR_LR_POSITIVE.format(self.lr))
    if self.batch_size < 1:
        raise ValueError(_ERR_BATCH_SIZE_RANGE.format(self.batch_size))
    if self.optimizer not in _VALID_OPTIMIZERS:
        raise ValueError(
            _ERR_UNKNOWN_OPTIMIZER.format(self.optimizer, sorted(_VALID_OPTIMIZERS))
        )
    if self.scheduler not in _VALID_SCHEDULERS:
        raise ValueError(
            _ERR_UNKNOWN_SCHEDULER.format(self.scheduler, sorted(_VALID_SCHEDULERS))
        )
    if self.grad_accum < 1:
        raise ValueError(
            _ERR_GRAD_ACCUM_RANGE.format(self.grad_accum)
        )
    if not 0.0 < self.ema_decay < 1.0:
        raise ValueError(
            _ERR_EMA_DECAY_RANGE.format(self.ema_decay)
        )

ExportConfig

corecv.api.model.ExportConfig(*, format='onnx', target_hardware='server', opset=_DEFAULT_OPSET, optimize=True, output_path=None, input_shape=(1, 3, *_DEFAULT_INPUT_SIZE), dynamic_axes=None) dataclass

Validated export configuration.

All fields are validated in __post_init__.

Attributes:

Name Type Description
format str

Export format. One of "onnx", "executorch", "both".

target_hardware str

Target hardware profile. "edge" applies activation rewrites (GELU -> ReLU, SiLU -> Hardswish); "server" skips rewrites.

opset int

ONNX opset version. Must be 17 or 18.

optimize bool

Apply additional graph optimisations (rewrites, layout folding, delegate passes). Default True.

output_path str | None

Explicit output file path. If None, a timestamped path is generated in a temporary directory.

input_shape tuple[int, ...]

Input tensor shape (B, C, H, W). Defaults to (1, 3, 224, 224).

dynamic_axes dict[str, dict[int, str]] | None

ONNX-style dynamic axes dictionary, e.g. {"input": {0: "batch", 2: "height", 3: "width"}}.

__post_init__()

Validate export configuration fields.

Source code in src/corecv/api/model.py
def __post_init__(self) -> None:
    """Validate export configuration fields."""
    if self.format not in _VALID_EXPORT_FORMATS:
        raise ValueError(
            _ERR_UNKNOWN_FORMAT.format(self.format, sorted(_VALID_EXPORT_FORMATS))
        )
    hw = self.target_hardware.lower()
    object.__setattr__(self, "target_hardware", hw)
    if hw not in _VALID_HARDWARE_TARGETS:
        raise ValueError(
            _ERR_UNKNOWN_HARDWARE.format(self.target_hardware, sorted(_VALID_HARDWARE_TARGETS))
        )
    if self.opset not in _VALID_OPSET_VERSIONS:
        raise ValueError(
            _ERR_INVALID_OPSET.format(self.opset, sorted(_VALID_OPSET_VERSIONS))
        )
    if len(self.input_shape) != _INPUT_SHAPE_NDIM:
        raise ValueError(
            _ERR_INPUT_SHAPE_DIMS.format(len(self.input_shape))
        )