Tasks

TaskAnalysis

TaskAnalysis(log)

Bases: BaseAnalysis

An object containing task analysis result. This should contain results including: result_df: dict A dataframe recording the details of all tests score: int Number of passed functionality (metabolic task) tests.

Parameters:

Name	Type	Description	Default
log		A dict storing parameters used to perform this analysis	required

Task

Task(
    should_fail: bool,
    in_mets: List[Dict[str, Any]],
    out_mets: List[Dict[str, Any]],
    knockout_input_flag: bool = True,
    knockout_output_flag: bool = True,
    ko_input_type: Literal["all", "organic"] = "all",
    ko_output_type: Literal["all", "organic"] = "all",
    compartment_parenthesis: str = "[{}]",
    met_id_str: str = "met_id",
    lower_bound_str: str = "lb",
    upper_bound_str: str = "ub",
    compartment_str: str = "compartment",
    system: str = None,
    subsystem: str = None,
    description: str = None,
    annotation: str = None,
)

Single metabolic task.

Attributes:

Name	Type	Description
system	str	The name of system that this task is in. Ex: Energy metabolism
subsystem	str	The name of subsystem that this task is in. Ex: OXYDATIVE PHOSPHORYLATION
description	str	A brief description about this task
should_fail	bool	Does this task should fail to pass or not.
in_mets	List[Dict[str, Any]]	The constraints(lower_bound, upper_bound) of influx metabolites Example : {'glc_D': {'lb': 0, 'ub': 1000, 'compartment': 'c'}, 'ATP': {'lb': 0, 'ub': 1000, 'compartment': 'c'}}
out_mets	List[Dict[str, Any]]	The constraints(lower_bound, upper_bound) of outflux metabolites Example : {'glc_D': {'lb': 0, 'ub': 1000, 'compartment': 'c'}, 'ATP': {'lb': 0, 'ub': 1000, 'compartment': 'c'}}

Initialize a metabolic task.

Parameters:

Name	Type	Description	Default
should_fail	bool	Indicates if the task is expected to fail (e.g., cannot carry flux).	required
in_mets	List[Dict[str, Any]]	List of dictionaries defining input metabolites and their bounds.	required
out_mets	List[Dict[str, Any]]	List of dictionaries defining output metabolites and their bounds.	required
knockout_input_flag	bool	Whether to knock out (set bounds to 0) boundary reactions for inputs not specified in in_mets. Default is True.	True
knockout_output_flag	bool	Whether to knock out boundary reactions for outputs not specified in out_mets. Default is True.	True
ko_input_type	Literal['all', 'organic']	Type of input boundary reactions to knock out ('all' or 'organic'). Default is "all".	'all'
ko_output_type	Literal['all', 'organic']	Type of output boundary reactions to knock out ('all' or 'organic'). Default is "all".	'all'
compartment_parenthesis	str	Format string for compartment identifiers (e.g., "[{}]"). Default is "[{}]".	'[{}]'
met_id_str	str	Dictionary key for metabolite ID within in_mets/out_mets. Default is "met_id".	'met_id'
lower_bound_str	str	Dictionary key for lower bound. Default is "lb".	'lb'
upper_bound_str	str	Dictionary key for upper bound. Default is "ub".	'ub'
compartment_str	str	Dictionary key for compartment ID. Default is "compartment".	'compartment'
system	str	Broad functional category of the task. Default is None.	None
subsystem	str	Specific metabolic pathway or subsystem. Default is None.	None
description	str	Textual description of the task. Default is None.	None
annotation	str	Additional annotations for the task. Default is None.	None

to_dict

to_dict() -> dict

Convert the Task object attributes to a dictionary.

mets_in_model

mets_in_model(model, all_mets_in_model=None) -> bool

A function to check if all the tasks' metabolites are in the model

Parameters:

Name	Type	Description	Default
model		Checked metabolic model	required
all_mets_in_model		All the metabolites' ID. Input to prevent to traverse the metabolite list in the model repeatedly.	None

Returns:

Name	Type	Description
are_in_model	bool	A bool indicates if the metabolites are all in the model

setup_support_flux_exp staticmethod

setup_support_flux_exp(model, rxn_fluxes: Dict[str, float])

Set up model objective and bounds based on supporting reaction fluxes.

Used internally for testing task sinks. Modifies the model in place.

Parameters:

Name	Type	Description	Default
model	Model	The metabolic model to modify.	required
rxn_fluxes	Dict[str, float]	A dictionary mapping reaction IDs to their flux values obtained from a previous task simulation (e.g., pFBA). Reaction bounds are adjusted slightly around these fluxes, and the objective is set to maximize flux through these reactions (weighted by sign).	required

assign

assign(
    model,
    all_mets_in_model=None,
    add_output_rxns=True,
    add_input_rxns=True,
    loose_input=False,
    loose_output=False,
    met_scaling_coefs=None,
)

Assign this task to a model and do the test

Parameters:

Name	Type	Description	Default
model		Assigned metabolic model	required
all_mets_in_model		All the metabolites in the model	None
loose_input		To loose the input constraints to the maximum (0, 1000)	False
loose_output		To loose the output constraints to the maximum (0, 1000)	False

Returns:

Type	Description
tuple	A tuple containing whether all metabolites exist, added dummy reactions, and objective reactions.

TaskContainer

TaskContainer(tasks: Optional[Dict[str, Task]] = None)

A container to hold and manage multiple Task objects.

Attributes:

Name	Type	Description
tasks	Dict[str, Task]	A dictionary where keys are task IDs (str) and values are Task objects.
ALLOW_BATCH_CHANGED_ATTR	List[str]	Class attribute listing Task attributes that can be modified in batch using set_all_mets_attr.

Initialize the TaskContainer.

Parameters:

Name	Type	Description	Default
tasks	Optional[Dict[str, Task]]	A dictionary of tasks to initialize the container with. Keys are task IDs, values are Task objects. Defaults to None, creating an empty container.	None

Raises:

Type	Description
TypeError	If tasks is provided and is not a dictionary.

items

items()

Return an iterator over the (task_id, Task) items.

subset

subset(items: List[str]) -> TaskContainer

Create a new TaskContainer containing only specified task IDs.

Parameters:

Name	Type	Description	Default
items	List[str]	A list of task IDs to include in the subset.	required

Returns:

Type	Description
TaskContainer	A new container with the subset of tasks.

load classmethod

load(
    file_path: Union[str, Path] = TASKS_FILE_PATH,
) -> TaskContainer

Load tasks from a JSON file.

Searches for the file first at the provided path, then in the default 'tasks' directory relative to the package root if the path is not absolute.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the JSON file containing task definitions. Defaults to the path defined by TASKS_FILE_PATH.	TASKS_FILE_PATH

Returns:

Type	Description
TaskContainer	A new container populated with tasks from the file.

save

save(file_path: Union[str, Path])

Save the tasks in the container to a JSON file.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path where the JSON file will be saved.	required

set_all_mets_attr

set_all_mets_attr(attr_name: str, new_value: Any)

Set a specific attribute for all Task objects within the container.

Only attributes listed in ALLOW_BATCH_CHANGED_ATTR can be modified.

Parameters:

Name	Type	Description	Default
attr_name	str	The name of the Task attribute to modify.	required
new_value	Any	The new value to assign to the attribute.	required

Raises:

Type	Description
AssertionError	If attr_name is not in ALLOW_BATCH_CHANGED_ATTR.

TaskHandler

TaskHandler(
    model: Model,
    tasks_path_or_container: Union[
        TaskContainer, str, Path
    ],
    model_compartment_parenthesis: str = "[{}]",
)

Handles the execution and analysis of metabolic tasks on a given model.

Attributes:

Name	Type	Description
model	Model	The metabolic model on which tasks will be tested.
tasks	TaskContainer	The container holding the metabolic tasks to be tested.

Initialize the TaskHandler.

Parameters:

Name	Type	Description	Default
model	Model	The metabolic model to use for task testing.	required
tasks_path_or_container	Union[TaskContainer, str, Path]	Either a TaskContainer object or a path to a JSON file defining the tasks.	required
model_compartment_parenthesis	str	The format string used for compartment identifiers in the model's metabolite IDs (e.g., "[{}]" for "met_c[c]"). Defaults to "[{}]".	'[{}]'

test_one_task

test_one_task(
    task,
    model,
    all_mets_in_model,
    method,
    method_kws,
    solver,
    fail_threshold: float,
    n_additional_path: int = 0,
    **kwargs
) -> dict

Test a single metabolic task on a given model context.

Applies task constraints (inputs, outputs, knockouts), runs flux analysis (e.g., pFBA), and determines if the task passes based on feasibility and expected outcome (should_fail). Optionally identifies supporting reactions.

Parameters:

Name	Type	Description	Default
task	Task	The metabolic task object to test.	required
model	Model	The model context (potentially modified with knockouts) to test the task on.	required
all_mets_in_model	List[str]	A list of all metabolite IDs present in the original model.	required
method	str	Flux analysis method to use (e.g., "pFBA", "FBA").	required
method_kws	Optional[Dict]	Keyword arguments for the chosen flux analysis method.	required
solver	str	The LP solver to use (e.g., "gurobi", "cplex").	required
fail_threshold	float	Flux threshold below which a reaction is considered inactive.	required
n_additional_path	int	Number of additional paths (minimal flux solutions) to find for identifying alternative supporting reactions. Default is 0.	0
**kwargs		Additional keyword arguments passed to task.assign.	{}

Returns:

Type	Description
dict	A dictionary containing results: 'Passed', 'Should fail', 'Missing mets', 'Status', 'Obj_value', 'Obj_rxns', 'system', 'subsystem', 'description', 'annotation', and optionally 'task_support_rxns' and 'task_support_rxn_fluxes'.

test_tasks

test_tasks(
    method="pFBA",
    method_kws=None,
    solver="gurobi",
    get_support_rxns=True,
    task_ids="all",
    verbosity=0,
    fail_threshold=1e-06,
    n_additional_path=0,
    met_scaling_coefs: Optional[Dict] = None,
    log: Optional[Dict] = None,
) -> TaskAnalysis

Test a set of metabolic tasks on the model.

Iterates through specified tasks, applies constraints, runs flux analysis, and aggregates results into a TaskAnalysis object.

Parameters:

Name	Type	Description	Default
method	str	Flux analysis method used for testing each task (e.g., "pFBA"). Default is "pFBA".	'pFBA'
method_kws	Optional[Dict]	Keyword arguments for the flux analysis method. Default is None.	None
solver	str	LP solver to use. Default is "gurobi".	'gurobi'
get_support_rxns	bool	If True, attempts to identify reactions supporting passed tasks by analyzing flux distributions and testing sinks. Default is True.	True
task_ids	Union[str, List[str]]	Specific task IDs to test. If "all" (default), tests all tasks in the container.	'all'
verbosity	int	Level of printed output (0: silent, 1: summary, 2: detailed). Default is 0.	0
fail_threshold	float	Flux threshold for determining reaction activity. Default is 1e-6.	1e-06
n_additional_path	int	Number of additional paths to find for supporting reactions. Default is 0.	0
met_scaling_coefs	Optional[Dict]	Dictionary mapping metabolite IDs to scaling coefficients, used if the model stoichiometry has been rescaled. Default is None.	None
log	Optional[Dict]	Additional information to store in the TaskAnalysis log. Default is None.	None

Returns:

Type	Description
TaskAnalysis	An object containing the aggregated results (DataFrame, score) and logs.

get_logger

get_logger(name: str) -> logging.Logger

Get a logger within the pipeGEM namespace.

Parameters:

Name	Type	Description	Default
name	str	The name of the logger, typically __name__ of the calling module.	required

Returns:

Type	Description
Logger

get_met_prod_task

get_met_prod_task(met_id: str, comp: str = 'c') -> Task

Create a simple metabolic task for producing a single metabolite.

This task defines the production of a specified metabolite from any available precursors in the model, with no specific inputs defined and boundary reactions generally open (unless knockout flags are changed later).

Parameters:

Name	Type	Description	Default
met_id	str	The metabolite's ID of the created Task	required
comp	str	The compartment of the metabolite	'c'

Returns:

Name	Type	Description
task	Task

table_to_container

table_to_container(
    df,
    met_id_map=None,
    id_name="ID",
    should_fail_name="SHOULD FAIL",
    desc_name="DESCRIPTION",
    in_met_name="IN",
    in_met_lb="IN LB",
    in_met_ub="IN UB",
    out_met_name="OUT",
    out_met_lb="OUT LB",
    out_met_ub="OUT UB",
    sys_name="SYSTEM",
    sub_sys_name="SUBSYSTEM",
    compartment_format: str = "\\[(.?)\\]",
) -> TaskContainer

Construct a TaskContainer from a pandas DataFrame.

Parses a DataFrame where rows define aspects of metabolic tasks (inputs, outputs, metadata) and converts it into a TaskContainer object. Assumes a specific table structure defined by the column name parameters.

Parameters:

Name	Type	Description	Default
df	DataFrame	The input DataFrame containing task definitions.	required
met_id_map	Optional[Dict]	A dictionary to map metabolite IDs found in the table to different IDs if needed. Default is None.	None
id_name	str	Column name for the unique Task ID. Default is "ID".	'ID'
should_fail_name	str	Column name indicating if the task should fail (boolean). Default is "SHOULD FAIL".	'SHOULD FAIL'
desc_name	str	Column name for the task description. Default is "DESCRIPTION".	'DESCRIPTION'
in_met_name	str	Column name for input metabolite IDs (including compartment). Default is "IN".	'IN'
in_met_lb	str	Column name for input metabolite lower bound. Default is "IN LB".	'IN LB'
in_met_ub	str	Column name for input metabolite upper bound. Default is "IN UB".	'IN UB'
out_met_name	str	Column name for output metabolite IDs (including compartment). Default is "OUT".	'OUT'
out_met_lb	str	Column name for output metabolite lower bound. Default is "OUT LB".	'OUT LB'
out_met_ub	str	Column name for output metabolite upper bound. Default is "OUT UB".	'OUT UB'
sys_name	str	Column name for the task system. Default is "SYSTEM".	'SYSTEM'
sub_sys_name	str	Column name for the task subsystem. Default is "SUBSYSTEM".	'SUBSYSTEM'
compartment_format	str	Regular expression to extract the compartment ID from metabolite strings (e.g., in in_met_name, out_met_name). Default captures a single character within square brackets: r"[(.?)]".	'\\[(.?)\\]'

Returns:

Type	Description
TaskContainer	A container populated with Task objects derived from the DataFrame.