Metadata-Version: 2.4
Name: motoko
Version: 0.1.4
Summary: Versatile Workflow Manager
Author-email: Guillaume Anciaux <guillaume.anciaux@epfl.ch>, Max Ludwig Hodapp <MaxLudwig.Hodapp@mcl.at>
License: GPL-3.0-or-later
Project-URL: homepage, https://gitlab.com/blackdynamite/motoko
Project-URL: repository, https://gitlab.com/blackdynamite/motoko
Project-URL: documentation, https://gitlab.com/blackdynamite/motoko
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: ase<4.0.0,>=3.24.0
Requires-Dist: blackdynamite>=1.3.9

# Motoko

![Motoko icon](icon.png)

Motoko is a workflow manager built on top of
[BlackDynamite](https://gitlab.com/blackdynamite/blackdynamite), a
Python-based tool for managing job submissions suited for HPC facilities in an
automated and highly parallelized fashion.

Full documentation can be found on [readthedocs](https://motoko.readthedocs.io/en/latest/)

## Installation

Install Motoko and dependencies:

```bash
pip install motoko
```

For installation in development mode:

```bash
pip install -e motoko_dir/
```

To run the local test workflow, do:

```bash
pytest
```

## BlackDynamite Studies

More concretely, Motoko manages workflows of interconnected
``BlackDynamite studies``. A BlackDynamite study is a parametrized calculation
that is possibly run many times with different inputs.

BlackDynamite parameterizes a calculation using ``jobs`` and ``runs``:

- A ``job`` is a calculation with a specific input.
- A ``run`` is an execution of a job. It carries metadata, the information
  needed to launch and track the calculation, and output quantities produced by
  the executable.

Several runs can
belong to the same job, for example when the same input is repeated after a
failure or evaluated with different software settings.

## Key Concepts

**Task Manager**

A wrapper around one BlackDynamite study.
A task is one pair of jobs and runs.
The Motoko task manager creates such job/run pairs and can also
select subsets of runs using constraints (e.g., selects only finished runs).

**Workflow**

The top-level object. It owns all task
managers, and coordinates their creation and execution.

**Orchestrator**

A Python module attached to a **Workflow**. It registers asynchronous routines
that the workflow executes following a dependency graph, or events triggering
actions when a certain condition becomes true.

## Workflow Directory Layout

A workflow directory contains `motoko.yaml`, an orchestrator Python file, and
one subdirectory per study:

```text
workflow/
  motoko.yaml
  orchestrator.py
  study1/
    bd.yaml
    launch.sh
    doIt.py
  study2/
    bd.yaml
    launch.sh
    doIt.py
```

Each study directory contains three files: `bd.yaml`, containing information
about the parametric space; `doIt.py`, containing the code that runs the
calculations; and `launch.sh`, which launches `doIt.py` with a given job
submission manager (e.g. Slurm).

Once the workflow has been started, Motoko saves the workflow state in
`workflow/.wf/`. BlackDynamite saves each study's state in
`workflow/study*/.bd/`. Directories containing the produced output of each run
are stored in `BD-study*-runs/`.

## `motoko.yaml`

`motoko.yaml` is the workflow input file. It declares the task-manager
directories and the Python function that will orchestrate them.

Minimal example:

```yaml
task_managers:
  task1:
  task2:

orchestrator: orchestrator.main
```

Detailed example:

```yaml
task_managers:
  prepare:
  solve:
    host: zeo://cluster_name:8010
  analyze:

aliases:
  post: analyze

orchestrator: orchestrator.main

generator: slurmCoat
slurm_options:
  - nodes=1
```

Fields:

- `task_managers`: Required mapping. Each key is both the task-manager name and
  the relative directory name containing that task's BlackDynamite study. Empty
  values are allowed. A task manager may define `host` to override the default
  local ZEO host.
- `orchestrator`: Required string in `module.function` form. Motoko loads
  `module.py` from the workflow directory and calls
  `function(workflow, **params)`.
- `aliases`: Optional mapping from alternative workflow attribute names to
  task-manager names. For example, `post: analyze` allows `workflow.post` to
  access the `analyze` task manager.
- `generator`: Optional BlackDynamite launcher generator (bash, slurm, PBS).
- `*_options`: Optional launcher-specific option lists. Motoko derives the key
  from the generator name with `generator.replace("Coat", "_options")` and
  appends the listed values to the launcher command.

## Python Orchestrator API

The orchestrator module usually exposes two functions:

```python
from motoko.workflow import event


def populate_arg_parser(parser):
    parser.add_argument("--inputs", "-i", type=float, required=True, nargs=2)


async def main(workflow, **params):
    # orchestrate the workflow here
```

`populate_arg_parser(parser)` is optional but useful for workflow-specific CLI
arguments. Motoko calls it when launching the orchestrator function from the
command line (see below).

`main(workflow, **params)` registers routines and actions. It should be async.
It normally ends by setting `workflow.finished = True` when some condition is
met.

### Asynchronous routines

Motoko orchestrators are asynchronous. This allows an orchestrator to express a
sequence of dependent routines directly in `main`.

Use this pattern when the workflow follows a dependency graph that can be
written as a sequence of awaited task submissions and selections:

```python
async def run_mult(workflow, inputs):
    return await workflow.mult.createTask(x=inputs)


async def run_add(workflow, mult_runs):
    created = []
    for run, job in mult_runs:
        created.extend(await workflow.add.createTask(x=run.y))
    return created


async def main(workflow, **params):
    mult_runs = await run_mult(workflow, params["inputs"])
    add_runs = await run_add(workflow, mult_runs)
    await workflow.norm.createTask(
        mult_ids=[run.id for run, job in add_runs],
    )
```

Figure: asynchronous routines called directly from `main`.

```mermaid
flowchart LR
    inputs[CLI inputs] --> mult[run_mult]
    mult --> add[run_add]
    add --> norm[create norm task]
```

`TaskManager.createTask(...)` returns an awaitable `RunList`. Awaiting it
commits the transaction, then waits until all created runs reach the
`FINISHED` state and returns the created `(run, job)` pairs:

```python
created = await workflow.add.createTask(x=1.5)
for run, job in created:
    print(run.id, job.id)
```

`TaskManager.select(...)` also returns an awaitable selection. Awaiting a
selection polls until at least one run matches the supplied BlackDynamite
constraints:

```python
finished = await workflow.mult.select("state = FINISHED")
```

### Registering Actions

Use actions for event-driven workflows. An action registered with
`add_action(...)` is triggered each time its event condition is met. In that
case, Motoko polls the condition and calls the action function when the
condition fires:

```python
workflow.add_action(event_name, task="__all__", event=..., f=...)
```

- `event_name`: Human-readable name stored in workflow logs.
- `task`: Task manager to watch. Use `"__all__"` to evaluate the event against
  every task manager.
- `event`: Condition that decides whether the action fires. It may be a
  BlackDynamite constraint string, a list of constraint strings, a `run, job`
  function, a `workflow, task_manager` function, or a no-argument function.
- `f`: Callback to run when the event fires. If the event returns a run
  selection, Motoko passes it as `runs=...`; otherwise the callback receives
  `workflow=...` and the runtime parameters.


Example actions:

```python
@event
async def spawn_init_tasks(workflow, **kwargs):
    await workflow.mult.createTask(x=kwargs["inputs"])


@event
async def spawn_add_tasks(runs=None, workflow=None, **kwargs):
    for run, job in runs:
        created = await workflow.add.createTask(x=run.y)
        run.state = "FORWARDED"
        run.dependencies = [f"add.{r.id}" for r, j in created]
```

Constraint-based event:

```python
workflow.add_action(
    "mult_finished",
    task="mult",
    event=["runs.id < 3", "state = FINISHED"],
    f=spawn_add_tasks,
)
```

Python condition event:

```python
def ready_for_norm(workflow, task_manager):
    if len(workflow.mult.select([])) != 2:
        return False
    if workflow.mult.select(["state != FORWARDED"]):
        return False
    return True


workflow.add_action("need_norm", event=ready_for_norm, f=spawn_norm_tasks)
```

### Workflow API Reference

- `Workflow(filename)`: Load `motoko.yaml`, create task manager objects, and
  record workflow paths.
- `workflow.create(validated=False)`: Reset `.wf/` and initialize all
  task-manager BlackDynamite studies.
- `workflow.start_launcher_daemons(args=None)`: Start BlackDynamite launcher
  daemons for all or selected task managers.
- `workflow.add_action(...)`: Register an event condition and callback.
- `workflow.add_error_handler(event="state = FAILED", f=...)`: Register a
  fail-fast action for failed runs.
- `workflow.execute(**params)`: Run the orchestrator and event polling loop.
- `workflow.get_runs(["add.1", "norm.3"])`: Resolve dependency references into
  persistent run objects grouped by task manager.
- `workflow.vars`: Persistent workflow variable namespace backed by
  `.wf/wf.db`.
- `workflow.<task_manager>`: Attribute access to task managers, for example
  `workflow.mult`.

### Task Manager API Reference

- `TaskManager.createTask(run_params=None, **job_params)`: Create one or more
  BlackDynamite jobs/runs. `job_params` are expanded through the study's
  BlackDynamite job schema and default `job_space`. `run_params` are stored on
  each run. Returns an awaitable `RunList` of `(run, job)` pairs.
- `await TaskManager.createTask(...)`: Wait until all created runs reach
  `FINISHED`, then return the created `(run, job)` pairs.
- `TaskManager.select(constraints=None)`: Return a lazy `TaskSelection`.
  Constraints use BlackDynamite syntax, such as `"state = FINISHED"` or
  `["runs.id < 3", "state = FINISHED"]`. When `workflow.run_name` is set,
  Motoko automatically adds a matching `run_name` constraint.
- `await TaskManager.select(...)`: Poll until the selection becomes non-empty.
- `TaskSelection.all(...)`: Build an awaitable condition that waits until all
  selected runs satisfy one of the supplied constraint sets.

## Launching a Motoko workflow

Motoko workflows can be run using the command line or within a Python script.

### Command Line Usage

Create or reset the BlackDynamite studies for a workflow:

```bash
motoko create workflow_dir
```

Start launcher daemons from inside the workflow directory:

```bash
cd workflow_dir
motoko launcher
```

Run the orchestrator in the foreground:

```bash
motoko orchestrator start --run_name test --inputs 2.1 3.2
```

Run the orchestrator detached with `zdaemon`:

```bash
motoko orchestrator start --detach --run_name test --inputs 2.1 3.2
```

Inspect workflow state:

```bash
motoko info
motoko info --verbose
motoko info --bd_study mult
```

Stop daemons:

```bash
motoko orchestrator stop
motoko kill
```

Clean BlackDynamite runs:

```bash
motoko clean
motoko clean --delete
```

### Python interface

The command line interface is a thin wrapper around the Python API. A workflow
can also be created and executed directly from Python:

```python
from motoko.workflow import Workflow


workflow = Workflow("motoko.yaml")
workflow.create()
workflow.start_launcher_daemons()
workflow.run_name = "test"
workflow.execute(inputs=[2.1, 3.1])
```

The `Workflow` constructor reads `motoko.yaml`, creates one `TaskManager` per
entry in `task_managers`, and exposes each manager as an attribute.
Tasks can also be created manually outside the orchestrator function:

```python
workflow = Workflow("motoko.yaml")
workflow.run_name = "manual"


async def submit_tasks():
    runs = await workflow.mult.createTask(x=[2.1, 3.1])
    finished = workflow.mult.select("state = FINISHED")
    return runs, finished
```

In a script, `workflow.run_name` must be set before creating tasks. Motoko
stores it on every created run and uses it to scope later selections, so
separate workflow executions do not accidentally interfere with one another's
runs.

Motoko loads `orchestrator.py` from the workflow directory and calls
`main(workflow, **params)`. Parameters come from CLI arguments and are also
accepted by `workflow.execute(**params)` when running from Python.

For scripts that only need to add or inspect tasks, use the task managers
directly:

```python
workflow = Workflow("motoko.yaml")
workflow.run_name = "inspection"

selected = workflow.norm.select(["state = FINISHED"])
for run, job in selected:
    print(run.id, run.state)
```

When using the Python API outside the CLI, the caller is responsible for
starting and stopping BlackDynamite launcher daemons if tasks should execute
automatically:

```python
subprocess.call("motoko kill", shell=True, cwd=workflow_dir)
```

## Troubleshooting

- `FATAL: not in a motoko directory (needs motoko.yaml)`: Several CLI commands
  expect to run from inside a workflow directory.
- No runs execute: Confirm `motoko launcher` is running for the relevant task
  managers.
- Selections return old or unexpected runs: Use a distinct `--run_name`. Motoko
  scopes selections by run name when it is set.
- Workflow never finishes: Ensure one action eventually sets
  `workflow.finished = True`.
- ZEO cache warnings: Stop daemons with `motoko kill`, recreate the workflow
  with `motoko create`, and rerun launchers.
