otx.cli.utils.hpo#

Utils for HPO with hpopt.

Functions

`get_best_hpo_weight`(hpo_dir, trial_id)	Get best model weight path of the HPO trial.
`run_hpo`(hpo_time_ratio, output, environment, ...)	Run HPO and load optimized hyper parameter and best HPO model weight.
`run_trial`(hp_config, report_func, ...)	Function to train a model given hyper parameters.

Classes

`HpoCallback`(report_func, metric, max_epoch, task)	Callback class to report score to HPO.
`HpoDataset`(fullset[, config, indices])	Wrapper class for DatasetEntity of dataset.
`HpoRunner`(environment, train_dataset_size, ...)	Class which is in charge of preparing and running HPO.
`TaskEnvironmentManager`(environment)	OTX environment utility class to set or get a value from environment class.
`TaskManager`(task_type)	Task utility class to give common interface from different task.
`Trainer`(hp_config, report_func, ...)	Class which prepares and trains a model given hyper parameters.

class otx.cli.utils.hpo.HpoCallback(report_func: Callable, metric: str, max_epoch: int, task)[source]#

Bases: UpdateProgressCallback

Callback class to report score to HPO.

Parameters:

report_func (Callable) – function to report score
metric (str) – metric name
max_epoch (int) – max_epoch
task – OTX train task

__call__(progress: int | float, score: float | None = None)[source]#: When callback is called, report a score to HPO algorithm.

class otx.cli.utils.hpo.HpoDataset(fullset, config: Dict[str, Any] | None = None, indices: List[int] | None = None)[source]#

Bases: object

Wrapper class for DatasetEntity of dataset. It’s used to make subset during HPO.

Parameters:

fullset – full dataset
config (Optional[Dict[str, Any]], optional) – hyper parameter trial config
indices (Optional[List[int]]) – dataset index. Defaults to None.

get_subset(subset: Subset)[source]#

Get subset according to subset_ratio if training dataset is requested.

Parameters:: subset (Subset) – which subset to get
Returns:: subset wrapped by HpoDataset
Return type:: HpoDataset

class otx.cli.utils.hpo.HpoRunner(environment: TaskEnvironment, train_dataset_size: int, val_dataset_size: int, hpo_workdir: str | Path, hpo_time_ratio: int = 4, progress_updater_callback: Callable[[int | float], None] | None = None)[source]#

Bases: object

Class which is in charge of preparing and running HPO.

Parameters:

environment (TaskEnvironment) – OTX environment
train_dataset_size (int) – train dataset size
val_dataset_size (int) – validation dataset size
hpo_workdir (Union[str, Path]) – work directory for HPO
hpo_time_ratio (int, optional) – time ratio to use for HPO compared to training time. Defaults to 4.
progress_updater_callback (Optional[Callable[[Union[int, float]], None]]) – callback to update progress

run_hpo(train_func: Callable, data_roots: Dict[str, Dict]) → Dict[str, Any] | None[source]#

Run HPO and provides optimized hyper parameters.

Parameters:

train_func (Callable) – training model function
data_roots (Dict[str, Dict]) – dataset path of each dataset type

Returns:

Optimized hyper parameters. If there is no best hyper parameter, return None.

Return type:

Union[Dict[str, Any], None]

class otx.cli.utils.hpo.TaskEnvironmentManager(environment: TaskEnvironment)[source]#

Bases: object

OTX environment utility class to set or get a value from environment class.

Parameters:: environment (TaskEnvironment) – OTX task environment

get_batch_size_name() → str[source]#

Get proper batch size name depending on task.

Returns:: batch size name
Return type:: str

get_dict_type_hyper_parameter() → Dict[str, Any][source]#

Get dictionary type hyper parmaeter of environment.

Returns:: dictionary type hyper parameter of environment
Return type:: Dict[str, Any]

get_max_epoch() → int[source]#

Get max epoch from environment.

Returns:: max epoch of environment
Return type:: int

get_model_template()[source]#: Get model template.

get_model_template_path() → str[source]#

Get model template path.

Returns:: path of model template
Return type:: str

get_new_model_entity(dataset=None) → ModelEntity[source]#

Get new model entity using environment.

Parameters:: dataset (Optional[DatasetEntity]) – OTX dataset
Returns:: new model entity
Return type:: ModelEntity

get_task() → TaskType[source]#

Get task type of environment.

Returns:: task type
Return type:: TaskType

get_train_task()[source]#

Get OTX train task instance.

Returns:: OTX train task instance
Return type:: OTX task

load_model_weight(model_weight_path: str, dataset: DatasetEntity)[source]#

Set model weight on environment to load the weight during training.

Parameters:

model_weight_path (str) – model weight to load during training
dataset (DatasetEntity) – dataset for training a model

resume_model_weight(model_weight_path: str, dataset: DatasetEntity)[source]#

Set model weight on environment to resume the weight during training.

Parameters:

model_weight_path (str) – model weight to resume during training
dataset (DatasetEntity) – dataset for training a model

save_initial_weight(save_path: Path | str) → bool[source]#

Save an initial model weight.

Parameters:: save_path (Union[str, Path]) – path to save initial model weight
Returns:: whether model weight is saved successfully
Return type:: bool

set_epoch(epoch: int)[source]#

Set epoch on environment.

Parameters:: epoch (int) – epoch to set

set_hyper_parameter_using_str_key(hyper_parameter: Dict[str, Any])[source]#

Set hyper parameter to environment using string key hyper_parameter.

Set hyper parameter to environment. Argument hyper_parameter is a dictionary which has string key. For example, hyper_parameter has a key “a.b.c”, then value is set at env_hp.a.b.c.

Parameters:: hyper_parameter (Dict[str, Any]) – hyper parameter to set which has a string format

property environment#: Environment property.

class otx.cli.utils.hpo.TaskManager(task_type: TaskType)[source]#

Bases: object

Task utility class to give common interface from different task.

Parameters:: task_type (TaskType) – otx task type

copy_weight(src: str | Path, det: str | Path)[source]#

Copy all model weights from work directory.

Parameters:

src (Union[str, Path]) – path where model weights are saved
det (Union[str, Path]) – path to save model weights

get_batch_size_name() → str[source]#

Give an proper batch size name depending on framework.

Returns:: batch size name
Return type:: str

get_epoch_name() → str[source]#

Give an proper epoch name depending on framework.

Returns:: epoch name
Return type:: str

get_latest_weight(workdir: str | Path) → str | None[source]#

Get latest model weight from all weights.

Parameters:: workdir (Union[str, Path]) – path where model weights are saved
Returns:: latest model weight path. If not found, than return None value.
Return type:: Optional[str]

is_anomaly_framework_task() → bool[source]#

Check taht task is run on anomalib.

Returns:: whether task is run on anomalib
Return type:: bool

is_cls_framework_task() → bool[source]#

Check that task is run on mmcls framework.

Returns:: whether task is run on mmcls
Return type:: bool

is_det_framework_task() → bool[source]#

Check that task is one of a task run on mmdet framework.

Returns:: whether task is run on mmdet
Return type:: bool

is_mmcv_framework_task() → bool[source]#

Check task is run on mmcv.

Returns:: whether task is run on mmcv
Return type:: bool

is_seg_framework_task() → bool[source]#

Check that task is run on mmseg framework.

Returns:: whether tasks is run on mmseg
Return type:: bool

property task_type#: Task_type property.

class otx.cli.utils.hpo.Trainer(hp_config: Dict[str, Any], report_func: Callable, model_template, data_roots: Dict[str, Dict], task_type: TaskType, hpo_workdir: str | Path, initial_weight_name: str, metric: str)[source]#

Bases: object

Class which prepares and trains a model given hyper parameters.

Parameters:

hp_config (Dict[str, Any]) – hyper parameter to use on training
report_func (Callable) – function to report score
model_template – model template
data_roots (Dict[str, Dict]) – dataset path of each dataset type
task_type (TaskType) – OTX task type
hpo_workdir (Union[str, Path]) – work directory for HPO
initial_weight_name (str) – initial model weight name for each trials to load
metric (str) – metric name

run()[source]#: Run each training of each trial with given hyper parameters.

otx.cli.utils.hpo.get_best_hpo_weight(hpo_dir: str | Path, trial_id: str | Path) → str | None[source]#

Get best model weight path of the HPO trial.

Parameters:

hpo_dir (Union[str, Path]) – HPO work directory path
trial_id (Union[str, Path]) – trial id

Returns:

best HPO model weight

Return type:

Optional[str]

otx.cli.utils.hpo.run_hpo(hpo_time_ratio: int, output: Path, environment: TaskEnvironment, dataset: DatasetEntity, data_roots: Dict[str, Dict], progress_updater_callback: Callable[[int | float], None] | None = None) → TaskEnvironment | None[source]#

Run HPO and load optimized hyper parameter and best HPO model weight.

Parameters:

hpo_time_ratio (int) – expected ratio of total time to run HPO to time taken for full fine-tuning
output (Path) – directory where HPO output is saved
environment (TaskEnvironment) – otx task environment
dataset (DatasetEntity) – dataset to use for training
data_roots (Dict[str, Dict]) – dataset path of each dataset type
progress_updater_callback (Optional[Callable[[Union[int, float]], None]]) – callback to update progress

otx.cli.utils.hpo.run_trial(hp_config: Dict[str, Any], report_func: Callable, model_template, data_roots: Dict[str, Dict], task_type: TaskType, hpo_workdir: str | Path, initial_weight_name: str, metric: str)[source]#

Function to train a model given hyper parameters.

Parameters:

hp_config (Dict[str, Any]) – hyper parameter to use on training
report_func (Callable) – function to report score
model_template – model template
data_roots (Dict[str, Dict]) – dataset path of each dataset type
task_type (TaskType) – OTX task type
hpo_workdir (Union[str, Path]) – work directory for HPO
initial_weight_name (str) – initial model weight name for each trials to load
metric (str) – metric name