class documentation

Logger wrapper for TensorBoard, WandB, and MLflow.

LuxonisTracker stores run metadata, initializes the selected logging integrations lazily, and keeps a local fallback cache for MLflow logs that fail transiently.

Static Method rank_zero_only Wrap a function so only processes with rank=0 execute it.
Method __init__ Create a tracker for one or more logging integrations.
Method close Finalize logging and save unsent logs locally.
Method log_hyperparams Log a hyperparameter dictionary.
Method log_image Log one image.
Method log_images Log multiple images.
Method log_matrix Log a matrix to the enabled logging services.
Method log_metric Log one scalar metric value.
Method log_metrics Log multiple scalar metrics.
Method log_stored_logs_to_mlflow Log any data stored in local_logs to MLflow.
Method log_to_mlflow Log to MLflow with retries.
Method save_logs_locally Save metrics, parameters, images, artifacts, and matrices locally.
Method store_log_locally Store log data locally if logging to MLflow fails.
Method upload_artifact Upload an artifact to the logging service.
Method upload_artifact_to_mlflow Upload an artifact specifically to MLflow.
Instance Variable is_mlflow Whether MLflow logging is enabled.
Instance Variable is_sweep Whether the current run belongs to a sweep.
Instance Variable is_tensorboard Whether TensorBoard logging is enabled.
Instance Variable is_wandb Whether WandB logging is enabled.
Instance Variable local_logs Locally cached MLflow payloads that will be retried or written to disk on close.
Instance Variable mlflow_initialized Whether MLflow initialization has succeeded.
Instance Variable mlflow_tracking_uri MLflow tracking URI used when MLflow logging is enabled.
Instance Variable project_id Project identifier used by WandB and MLflow.
Instance Variable project_name Project name used by WandB and MLflow.
Instance Variable rank Process rank. Only rank 0 writes through rank-gated logging methods.
Instance Variable run_directory Directory for local run artifacts.
Instance Variable run_id MLflow run identifier, used to resume an existing run.
Instance Variable run_name Name of the current run.
Instance Variable save_directory Root directory where local run outputs are stored.
Instance Variable wandb_entity WandB entity used for logging.
Property experiment Creates new experiments or returns active ones if already created.
Property name Run name.
Property version Tracker version.
Method _get_latest_run_name Return the most recently created run name.
Method _get_next_run_number Return the number ID for the next run.
Method _get_run_name Generate a new run name.
Instance Variable _experiment Undocumented
@staticmethod
def rank_zero_only(fn: Callable) -> Callable:

Wrap a function so only processes with rank=0 execute it.

def __init__(self, project_name: str | None = None, project_id: str | None = None, run_name: str | None = None, run_id: str | None = None, save_directory: PathType = 'output', is_tensorboard: bool = False, is_wandb: bool = False, is_mlflow: bool = False, is_sweep: bool = False, wandb_entity: str | None = None, mlflow_tracking_uri: str | None = None, rank: int = 0):

Create a tracker for one or more logging integrations.

Parameters
project_name:str | NoneProject name used for WandB and MLflow.
project_id:str | NoneProject ID used for WandB and MLflow.
run_name:str | NoneRun name. If omitted, rank 0 generates a new name and other ranks use the latest run name.
run_id:str | NoneMLflow run ID used to continue a previous run.
save_directory:PathTypeDirectory where local outputs are saved.
is_tensorboard:boolWhether to use TensorBoard logging.
is_wandb:boolWhether to use WandB logging.
is_mlflow:boolWhether to use MLflow logging.
is_sweep:boolWhether the current run is part of a sweep.
wandb_entity:str | NoneWandB entity to use.
mlflow_tracking_uri:str | NoneMLflow tracking URI to use.
rank:intProcess rank used in distributed training.
Raises
ValueErrorIf WandB or MLflow is enabled but neither project_name nor project_id is provided.
ValueErrorIf WandB is enabled without wandb_entity.
ValueErrorIf MLflow is enabled without mlflow_tracking_uri.
ValueErrorIf no logging integration is enabled.
def close(self):

Finalize logging and save unsent logs locally.

def log_hyperparams(self, params: dict[str, str | bool | int | float | None]):

Log a hyperparameter dictionary.

Parameters
params:dict[str, str | bool | int | float | None]Hyperparameter key-value pairs.
def log_image(self, name: str, img: np.ndarray, step: int):

Log one image.

Note

step is omitted when logging with WandB to avoid problems with inconsistent incrementation.

Parameters
name:strImage caption. For MLflow, this should include a slash-separated base path and image caption.
img:np.ndarrayImage data of shape (H, W, C).
step:intCurrent step.
def log_images(self, imgs: dict[str, np.ndarray], step: int):

Log multiple images.

Parameters
imgs:dict[str, np.ndarray]Mapping from image captions to image data of shape (H, W, C).
step:intCurrent step.
def log_matrix(self, matrix: np.ndarray, name: str, step: int, extra_data: dict | None = None):

Log a matrix to the enabled logging services.

Parameters
matrix:np.ndarrayMatrix to log, usually of shape (M, N).
name:strName used for the matrix artifact.
step:intCurrent step.
extra_data:dict | NoneOptional dictionary of additional data to include in the logged matrix artifact.
def log_metric(self, name: str, value: float, step: int):

Log one scalar metric value.

Note

step is omitted when logging with WandB to avoid problems with inconsistent incrementation.

Parameters
name:strMetric name.
value:floatMetric value.
step:intCurrent step.
def log_metrics(self, metrics: dict[str, float], step: int):

Log multiple scalar metrics.

Parameters
metrics:dict[str, float]Metric key-value pairs.
step:intCurrent step.
def log_stored_logs_to_mlflow(self):

Log any data stored in local_logs to MLflow.

def log_to_mlflow(self, log_fn: Callable, *args, **kwargs):

Log to MLflow with retries.

Logs locally if failures persist.

def save_logs_locally(self):

Save metrics, parameters, images, artifacts, and matrices locally.

def store_log_locally(self, log_fn: Callable, *args, **kwargs):

Store log data locally if logging to MLflow fails.

def upload_artifact(self, path: PathType, name: str | None = None, typ: str = 'artifact'):

Upload an artifact to the logging service.

Parameters
path:PathTypePath to the artifact.
name:str | NoneArtifact name. If None, uses the file stem for WandB and the file name for MLflow.
typ:strThe type of the artifact. Only used for WandB.
def upload_artifact_to_mlflow(self, path: PathType, name: str | None = None):

Upload an artifact specifically to MLflow.

Parameters
path:PathTypePath to the artifact.
name:str | NoneArtifact name. If None, uses the file name.
is_mlflow =

Whether MLflow logging is enabled.

is_sweep =

Whether the current run belongs to a sweep.

is_tensorboard =

Whether TensorBoard logging is enabled.

is_wandb =

Whether WandB logging is enabled.

local_logs: dict =

Locally cached MLflow payloads that will be retried or written to disk on close.

mlflow_initialized: bool =

Whether MLflow initialization has succeeded.

mlflow_tracking_uri =

MLflow tracking URI used when MLflow logging is enabled.

project_id =

Project identifier used by WandB and MLflow.

project_name =

Project name used by WandB and MLflow.

rank =

Process rank. Only rank 0 writes through rank-gated logging methods.

run_directory =

Directory for local run artifacts.

run_id =

MLflow run identifier, used to resume an existing run.

run_name =

Name of the current run.

save_directory =

Root directory where local run outputs are stored.

wandb_entity =

WandB entity used for logging.

@property
@rank_zero_only
experiment: dict[Literal['tensorboard', 'wandb', 'mlflow'], Any] =

Creates new experiments or returns active ones if already created.

@property
name: str =

Run name.

Returns
Current run name.
@property
version: int =

Tracker version.

Returns
Version number 1.
def _get_latest_run_name(self) -> str:

Return the most recently created run name.

def _get_next_run_number(self) -> int:

Return the number ID for the next run.

def _get_run_name(self) -> str:

Generate a new run name.

_experiment =

Undocumented