Skip to content

tasks

Task

A task offers a light-touch way for users to specify any arbitrary downstream task that they want to run on a dataset.

Parameters:

Name Type Description Default
name str

The name of the task.

 required
run Callable

The function to run.

 required
supports_fairness bool

Whether the task supports fairness evaluation.

 False
target str

The target column name for fairness evaluation (required if supports_fairness=True).

 None
description str

The description of the task.

 ''
supports_aequitas bool

Deprecated alias for supports_fairness (for backward compatibility).

 None
Source code in src/nhssynth/modules/evaluation/tasks.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class Task:
    """
    A task offers a light-touch way for users to specify any arbitrary downstream task that they want to run on a dataset.

    Args:
        name: The name of the task.
        run: The function to run.
        supports_fairness: Whether the task supports fairness evaluation.
        target: The target column name for fairness evaluation (required if supports_fairness=True).
        description: The description of the task.
        supports_aequitas: Deprecated alias for supports_fairness (for backward compatibility).
    """

    def __init__(
        self,
        name: str,
        run: Callable,
        supports_fairness: bool = False,
        target: str = None,
        description: str = "",
        supports_aequitas: bool = None,  # backward compatibility
    ):
        self._name: str = name
        self._run: Callable = run
        # Support backward compatibility with supports_aequitas
        if supports_aequitas is not None:
            supports_fairness = supports_aequitas
        self._supports_fairness: bool = supports_fairness
        self._target: str = target
        self._description: str = description
        if supports_fairness and target is None:
            raise ValueError(f"Task '{name}' supports fairness evaluation but no target column specified.")

    @property
    def name(self) -> str:
        return self._name

    @property
    def description(self) -> str:
        return self._description

    @property
    def supports_fairness(self) -> bool:
        return self._supports_fairness

    @property
    def supports_aequitas(self) -> bool:
        """Deprecated: Use supports_fairness instead."""
        return self._supports_fairness

    @property
    def target(self) -> str:
        return self._target

    def __str__(self) -> str:
        return f"{self.name}: {self.description}" if self.description else self.name

    def __repr__(self) -> str:
        return str([self.name, self.run, self.supports_fairness, self.target, self.description])

    def run(self, *args, **kwargs):
        return self._run(*args, **kwargs)

supports_aequitas property

Deprecated: Use supports_fairness instead.

get_tasks(fn_dataset, tasks_root)

Searches for and imports all tasks in the tasks directory for a given dataset. Uses importlib to extract the task from the file.

Parameters:

Name Type Description Default
fn_dataset str

The name of the dataset.

 required
tasks_root str

The root directory for downstream tasks.

 required

Returns:

Type Description
list[Task]

A list of tasks.
Source code in src/nhssynth/modules/evaluation/tasks.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_tasks(
    fn_dataset: str,
    tasks_root: str,
) -> list[Task]:
    """
    Searches for and imports all tasks in the tasks directory for a given dataset.
    Uses `importlib` to extract the task from the file.

    Args:
        fn_dataset: The name of the dataset.
        tasks_root: The root directory for downstream tasks.

    Returns:
        A list of tasks.
    """
    tasks_dir = Path(tasks_root) / fn_dataset
    assert (
        tasks_dir.exists()
    ), f"Downstream tasks directory does not exist ({tasks_dir}), NB there should be a directory in TASKS_DIR with the same name as the dataset."
    tasks = []
    for task_path in tasks_dir.iterdir():
        if task_path.name.startswith((".", "__")):
            continue
        assert task_path.suffix == ".py", f"Downstream task file must be a python file ({task_path.name})"
        spec = importlib.util.spec_from_file_location(
            "nhssynth_task_" + task_path.name, os.getcwd() + "/" + str(task_path)
        )
        task_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(task_module)
        tasks.append(task_module.task)
    return tasks