RUFAS.task_manager module#

RUFAS.task_manager.MINIMUM_PYTHON_VERSION = <Version('3.12')>#: These constants define the minimum and maximum integers that can be passed to Numpy’s random.seed method.

class RUFAS.task_manager.TaskType(*values)#

Bases: Enum

Enum for different task types handled by TaskManager.

HERD_INITIALIZATION = 'Herd Initialization'#

SIMULATION_SINGLE_RUN = 'A single simulation run'#

SIMULATION_MULTI_RUN = 'Multiple simulation with different random seeds'#

SENSITIVITY_ANALYSIS = 'Run sensitivity analysis'#

INPUT_DATA_AUDIT = 'Validates input data and saves metadata properties as CSV'#

END_TO_END_TESTING = 'Run e2e testing'#

POST_PROCESSING = 'Bypass simulation engine and directly run Output Manager'#

COMPARE_METADATA_PROPERTIES = 'Compares 2 metadata properties files and saves the differences in a .txt file'#

DATA_COLLECTION_APP_UPDATE = 'Updates the schema and interface of the Data Collection App'#

UPDATE_E2E_TEST_RESULTS = 'Updates end-to-end expected test results with new actual results'#

static from_string(input_str: str) → TaskType#: Converts a string to a TaskType enum.

is_multi_run() → bool#: Checks if the task type involves multiple runs.

class RUFAS.task_manager.TaskManager#

Bases: object

Manager class for handling tasks related to simulations and analyses.

INPUT_DATA_CSV_WORKING_FOLDER = PosixPath('output/saved_pool_working_folder')#

__init__() → None#

start(metadata_path: Path, verbosity: LogVerbosity, exclude_info_maps: bool, output_directory: Path, logs_directory: Path, clear_output_directory: bool, produce_graphics: bool, suppress_log_files: bool, metadata_depth_limit: int) → None#

Initializes and starts the task management process.

Parameters#

metadata_pathPath: Path to the metadata file that contains task management inputs.
verbosityLogVerbosity: Level of verbosity for logging.
exclude_info_mapsbool: Flag to exclude information maps.
output_directoryPath: Path to the directory where outputs will be saved.
logs_directoryPath: Path to the directory where logs from the Task Manager will be saved.
clear_output_directorybool: Whether to clear the output directory.
produce_graphicsbool: Whether to produce graphics.
suppress_log_filesbool: Whether to write logs from the Task Manager to output files.
metadata_depth_limitint: Override value for maximum metadata properties depth set in Input Manager.

get_rufas_version() → str#

Returns the current version of RUFAS.

Returns#

str: Version of RUFAS or “Unknown” if the version of Python version earlier than 3.12.

check_python_version() → None#: Checks if the Python version meets version range set in pyproject.toml.

_parse_input_tasks() → Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]#

Parses input tasks into single and multiple run tasks.

Returns#

Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: Parsed single run and multi-run task arguments.

_expand_multi_runs_to_single_runs(multi_run_args: List[Dict[str, Any]]) → List[Dict[str, Any]]#

Expands multi-run tasks into single-run tasks for execution.

Parameters#

multi_run_argsList[Dict[str, Any]]: List of multi-run task arguments.

Returns#

List[Dict[str, Any]]: Expanded list of single-run tasks.

_expand_simulation_multi_run_args(multi_run_args: Dict[str, Any]) → List[Dict[str, Any]]#

_expand_sensitivity_analysis_args(multi_run_args: Dict[str, Any]) → List[Dict[str, Any]]#: Expands sensitivity analysis multi-run tasks into single-run tasks.

_run_tasks(single_run_args: List[Dict[str, Any]], produce_graphics: bool, metadata_depth_limit: int, workers: int) → None#: Runs the tasks based on the provided arguments.

static call_handler(handler: Callable[[...], None], args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_graphics: bool, should_flush_im_pool: bool) → None#: Wrapper function to call the function map with each of its arguments.

static task(args: Dict[str, Any], produce_graphics: bool, workers: int, metadata_depth_limit: int | None) → str | None#: Executes a single task with specified arguments.

static handle_herd_initializaition(args: Dict[str, Any], output_manager: OutputManager) → None#: Handles initialization of the herd based on specified arguments.

static handle_single_simulation_run(args: Dict[str, Any], output_manager: OutputManager) → None#: Conducts a single simulation run based on provided arguments.

static _handle_end_to_end_testing(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: str, produce_graphics: bool, should_flush_im_pool: bool) → None#: Runs end-to-end testing routine.

static _handle_update_e2e_test_results(args: dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: str, produce_graphics: bool, should_flush_im_pool: bool) → None#: Generates a new set of end-to-end expected test results.

static handle_input_data_audit(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, eager_termination: bool) → bool#: Validates input data saves metadata properties to CSV.

static handle_post_processing(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: str, should_flush_im_pool: bool, produce_graphics: bool = False, save_results: bool = False, load_pool_from_file: bool = False, export_input_data_to_csv: bool = False) → None#

Handles post-processing tasks based on specified arguments.

Parameters#

argsDict[str, Any]: Arguments for post-processing.
input_managerInputManager: Manager to handle input processing.
output_managerOutputManager: Manager to handle output logging and errors.
task_id: str: The ID that Task Manager has assigned to this task.
produce_graphicsbool: Whether to produce graphics during post-processing.
save_resultsbool: Whether to save results after processing.
load_pool_from_filebool: Whether to load data pool from file.
export_input_data_to_csv: bool: Whether to export the input data to a CSV file.
should_flush_im_pool: bool: Whether to flush the input manager pool.

static set_random_seed(random_seed: int | None, output_manager: OutputManager) → None#: Sets the random seed for the task run.

static _handle_input_data_audit_tasks(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_grahics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to metadata property comparison.

static _handle_compare_metadata_properties_tasks(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_grahics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to metadata property comparison.

static _handle_herd_init_tasks(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_grahics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to herd initialization.

static _handle_simulation_engine_run_tasks(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_graphics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to simulation run.

static _handle_postprocessing_tasks(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_graphics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to postprocessing.

static _handle_data_collection_app_update(args: Dict[str, Any], input_manager: InputManager, output_manager: OutputManager, task_id: Any, produce_graphics: bool, should_flush_im_pool: bool) → None#: Handler for all methods related to updating the Data Collection App.