pcntoolkit.util.runner#

Classes#

Runner

Initialize the runner.

Functions#

load_and_execute(args)

Load a callable and data from a pickle file and execute it.

Module Contents#

class Runner(parallelize: bool = False, job_type: Literal['torque', 'slurm'] = 'slurm', n_batches: int | None = None, batch_size: int | None = None, n_cores: int = 1, time_limit: str | int = '00:05:00', memory: str = '5GB', max_retries: int = 3, environment: str | None = None, cross_validate: bool = False, cv_folds: int = 5, preamble: str = 'module load anaconda3', log_dir: str | None = None, temp_dir: str | None = None)#

Initialize the runner.

Parameters:

  • parallelize (bool, optional) – Whether to parallelize the jobs.

  • job_type (Literal[``”torque”, ``"slurm"], optional) – The type of job to use.

  • n_batches (int, optional) – The number of jobs to run in parallel.

  • n_cores (int, optional) – The number of cores to use for each job.

  • time_limit (str | int, optional) – The time limit for each job.

  • memory (str, optional) – The memory to use for each job.

  • max_retries (int, optional) – The maximum number of retries for each job.

  • environment (str, optional) – The environment to use for each job.

  • cross_validate (bool, optional) – Whether to cross-validate the model.

  • cv_folds (int, optional) – The number of folds to use for cross-validation.

  • preamble (str, optional) – The preamble to use for each job.

  • log_dir (str, optional) – The directory to save the logs to.

  • temp_dir (str, optional) – The directory to save the temporary files to.

check_job_status(job_name: str) → tuple[bool, bool, str | None]#

Check if a job has failed by looking for success file.

Returns:

(is_running, finished_with_error, error_message) If job is still running, returns (True, False, None) If job finished successfully, returns (False, False, None) If job failed, returns (False, True, error_message)

Return type:

tuple[bool, bool, Optional[str]]

check_jobs_status() → tuple[Dict[str, str], Dict[str, str], Dict[str, str]]#

Check all jobs in active_job_ids for errors.

Returns:

A tuple containing: - A dictionary mapping job names to job IDs for running jobs - A dictionary mapping job names to error messages for failed jobs - A dictionary mapping job names to job IDs for finished jobs

Return type:

tuple[Dict[str, str], Dict[str, str], Dict[str, str]]

create_temp_and_log_dir()#
extend(model: pcntoolkit.normative_model.NormativeModel, data: pcntoolkit.dataio.norm_data.NormData, save_dir: str | None = None, observe: bool = True, **kwargs) → pcntoolkit.normative_model.NormativeModel | None#

Extend a normative model on a dataset.

Parameters:

  • model (NormativeModel) – The normative model to extend.

  • data (NormData) – The data to extend the model on.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish and then load the model into the model object. If false, the function will dispatch the jobs and return.

Returns:

The extended model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormativeModel | None

extend_predict(model: pcntoolkit.normative_model.NormativeModel, fit_data: pcntoolkit.dataio.norm_data.NormData, predict_data: pcntoolkit.dataio.norm_data.NormData | None = None, save_dir: str | None = None, observe: bool = True, **kwargs) → pcntoolkit.normative_model.NormativeModel | None#

Extend a normative model on a dataset and predict on another dataset.

Parameters:

  • model (NormativeModel) – The normative model to extend.

  • fit_data (NormData) – The data to extend the model on.

  • predict_data (Optional[NormData], optional) – The data to predict on. Can be None if cross-validation is used.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – If false, the function will dispatch the jobs and return.

Returns:

The extended model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormativeModel | None

fit(model: pcntoolkit.normative_model.NormativeModel, data: pcntoolkit.dataio.norm_data.NormData, save_dir: str | None = None, observe: bool = True) → pcntoolkit.normative_model.NormativeModel | None#

Fit a normative model on a dataset.

Parameters:

  • model (NormBase) – The normative model to fit.

  • data (NormData) – The data to fit the model on.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish and then load the model into the model object. If false, the function will dispatch the jobs and return. In that case, the model will not be loaded into the model object, it will have to be loaded manually using the load function when the jobs are done.

Returns:

The fitted model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormativeModel | None

fit_predict(model: pcntoolkit.normative_model.NormativeModel, fit_data: pcntoolkit.dataio.norm_data.NormData, predict_data: pcntoolkit.dataio.norm_data.NormData | None = None, save_dir: str | None = None, observe: bool = True) → pcntoolkit.normative_model.NormativeModel | None#

Fit a normative model on a dataset and predict on another dataset.

Parameters:

  • model (NormativeModel) – The normative model to fit.

  • fit_data (NormData) – The data to fit the model on.

  • predict_data (Optional[NormData], optional) – The data to predict on. Can be None if cross-validation is used.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish, then load the model into the model object If false, the function will dispatch the jobs and return. In that case, the model will not be loaded into the model object, it will have to be loaded manually using the load function when the jobs are done.

Returns:

The fitted and model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormativeModel | None

classmethod from_args(args: dict) → Runner#
get_all_job_file_paths(job_name)#
get_data_path(job_name)#
get_extend_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str, **kwargs) → Callable#
get_extend_predict_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str, **kwargs) → Callable#
get_fit_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str) → Callable#

Returns a callable that fits a model on a chunk of data

get_fit_predict_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str) → Callable#

Returns a callable that fits a model on a chunk of data and predicts on another chunk of data

get_predict_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str) → Callable#

Loads each fold model and predicts on the corresponding fold of data. Model n is used to predict on fold n.

get_python_callable_path(job_name)#
get_transfer_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str, **kwargs) → Callable#

Returns a callable that transfers a model on a chunk of data

get_transfer_predict_chunk_fn(model: pcntoolkit.normative_model.NormativeModel, save_dir: str, **kwargs) → Callable#
load_data(data_source: pcntoolkit.dataio.norm_data.NormData, fold_index: int | None = 0) → None#
classmethod load_from_state(runner_file: str) → Runner#

Load a runner from a saved state.

Parameters:

runner_file (str) – Path to the runner state file

Returns:

A runner instance with the saved state

Return type:

Runner

load_model(fold_index: int | None = 0, into: pcntoolkit.normative_model.NormativeModel | None = None) → pcntoolkit.normative_model.NormativeModel#
predict(model: pcntoolkit.normative_model.NormativeModel, data: pcntoolkit.dataio.norm_data.NormData, save_dir: str | None = None, observe: bool = True) → pcntoolkit.normative_model.NormativeModel | None#

Predict on a dataset.

Parameters:

  • model (NormativeModel) – The normative model to predict on.

  • data (NormData) – The data to predict on.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish.

Return type:

None. If you want to load the model, use the runner.load_model function.

re_submit_failed_jobs(observe: bool = True) → None#
register_fold_indices(save_dir: str, i_fold: int, indices: tuple[int, int])#
save() → None#

Save the runner state to a JSON file in the save directory.

save_callable_and_data(job_name: int | str, fn: Callable, chunk: tuple[pcntoolkit.dataio.norm_data.NormData] | tuple[pcntoolkit.dataio.norm_data.NormData, pcntoolkit.dataio.norm_data.NormData | None]) → tuple[str, str]#
set_task_id(task_name: str, model: pcntoolkit.normative_model.NormativeModel, data: pcntoolkit.dataio.norm_data.NormData)#
submit_jobs(fn: Callable, first_data_source: pcntoolkit.dataio.norm_data.NormData, second_data_source: pcntoolkit.dataio.norm_data.NormData | None = None, mode: Literal['unary', 'binary'] = 'unary') → None#

Submit jobs to the job scheduler.

The predict_data argument is optional, and if it is not provided, None is passed to the function.

Parameters:

  • fn (Callable) – Function to call. It should take two arguments.

  • fit_data (NormData) – Data to fit the model on

  • predict_data (Optional[NormData], optional) – Data to predict on, by default None

transfer(model: pcntoolkit.normative_model.NormativeModel, data: pcntoolkit.dataio.norm_data.NormData, save_dir: str | None = None, observe: bool = True, **kwargs) → pcntoolkit.normative_model.NormativeModel | None#

Transfer a normative model to a new dataset.

Parameters:

  • model (NormativeModel) – The normative model to transfer.

  • data (NormData) – The data to transfer the model to.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish and then return the transfered model. If false, the function will dispatch the jobs and return.

Returns:

The transfered model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormBase | None

transfer_predict(model: pcntoolkit.normative_model.NormativeModel, fit_data: pcntoolkit.dataio.norm_data.NormData, predict_data: pcntoolkit.dataio.norm_data.NormData | None = None, save_dir: str | None = None, observe: bool = True, **kwargs) → pcntoolkit.normative_model.NormativeModel | None#

Transfer a normative model to a new dataset and predict on another dataset.

Parameters:

  • model (NormativeModel) – The normative model to transfer.

  • fit_data (NormData) – The data to transfer the model to.

  • predict_data (Optional[NormData], optional) – The data to predict on. Can be None if cross-validation is used.

  • save_dir (Optional[str], optional) – The directory to save the model to. If None, the model will be saved in the model’s save directory.

  • observe (bool, optional) – Whether to observe the jobs. If true, the function will wait for the jobs to finish and then load the model into the model object. If false, the function will dispatch the jobs and return.

Returns:

The transfered model. If observe is true, the function will wait for the jobs to finish and return the model object. If observe is false, the function will return None.

Return type:

NormBase | No   ne

wait_or_finish(observe: bool, into: pcntoolkit.normative_model.NormativeModel | None = None, *data_sources) → pcntoolkit.normative_model.NormativeModel | None#
wrap_in_job(job_name, python_callable_path, data_path)#
wrap_in_slurm_job(job_name: int | str, python_callable_path: str, data_path: str) → list[str]#
wrap_in_torque_job(job_name: int | str, python_callable_path: str, data_path: str) → list[str]#
active_jobs: Dict[str, str]#
batch_size: int | None = 2#
cross_validate: bool = False#
cv_folds: int = 5#
environment: str = None#
failed_jobs: Dict[str, str]#
job_commands: Dict[str, list[str]]#
job_observer = None#
job_type: str = 'local'#
log_dir: str = ''#
max_retries: int = 3#
memory: str = '5gb'#
n_batches: int | None = None#
n_cores: int = 1#
parallelize: bool = True#
preamble: str = 'module load anaconda3'#
save_dir = ''#
task_id = ''#
temp_dir: str = ''#
time_limit_seconds: int = 300#
time_limit_str: str | int = '00:05:00'#
unique_log_dir = ''#
unique_temp_dir = ''#
load_and_execute(args)#

Load a callable and data from a pickle file and execute it.

Parameters:

args (list[str]) – A list of arguments. The first argument is the path to the callable. The second argument is the path to the data. The third argument is the max number of retries.