Metadata-Version: 2.4
Name: modak
Version: 0.1.0
Summary: A simple, opinionated task manager
Author-email: Dene Hoffman <denehoffman@me.com>
Project-URL: Homepage, https://github.com/denehoffman/textdraw
Project-URL: Repository, https://github.com/denehoffman/textdraw
Project-URL: Issues, https://github.com/denehoffman/textdraw/issues
Keywords: task,job,scheduler,monitor
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE-MIT
License-File: LICENSE-APACHE
Requires-Dist: click>=8.2.0
Requires-Dist: numpy>=2.2.5
Requires-Dist: textdraw>=0.1.2
Requires-Dist: textual>=3.2.0
Requires-Dist: tzlocal>=5.3.1
Dynamic: license-file

<!-- markdownlint-disable MD033 MD041 -->
<p align="center">
  <h1 align="center">modak</h1>
</p>
<p align="center">
    <img alt="GitHub Release" src="https://img.shields.io/github/v/release/denehoffman/modak?style=for-the-badge&logo=github"></a>
  <a href="https://github.com/denehoffman/modak/commits/main/" alt="Latest Commits">
    <img alt="GitHub last commit" src="https://img.shields.io/github/last-commit/denehoffman/modak?style=for-the-badge&logo=github"></a>
  <a href="LICENSE-APACHE" alt="License">
    <img alt="GitHub License" src="https://img.shields.io/github/license/denehoffman/modak?style=for-the-badge"></a>
  <a href="https://pypi.org/project/modak/" alt="View project on PyPI">
  <img alt="PyPI - Version" src="https://img.shields.io/pypi/v/modak?style=for-the-badge&logo=python&logoColor=yellow&labelColor=blue"></a>
</p>

`modak` is a simple-to-use, opinionated task queue system with dependency
management, resource allocation, and isolation control. Tasks are run
respecting topological dependencies, resource limits, and optional isolation.

This library only has two classes, `Task`s, which are an abstract class with a
single method to override, `run(self) -> None`, and a `TaskQueue` which manages
the execution order. Additionally, `modak` comes with a task monitor TUI which
can be invoked with the `modak` shell command.

By default, `modak` scripts will create a state file and logs in a directory
called `.modak` in the current working directory. This can be changed by setting
the `MODAK_PATH` environment variable. The `modak` CLI also supports an optional
`-c/--cwd <PATH>` argument to set the working directory containing the `.modak` file.

## Features

- Topological task scheduling
- Persistent state and log files
- Resource-aware execution
- Isolated task handling
- Skipping of previously completed tasks based on input/output hashes

## Installation

```shell
pip install modak
```

Or with `uv`:

```shell
pip install modak
```

## Examples

### A simple chain of tasks

```python
from modak import Task, TaskQueue

class PrintTask(Task):
    def run(self):
        self.log_info(f"Running {self.name}")

t1 = PrintTask(name="task1")
t2 = PrintTask(name="task2", inputs=[t1])
t3 = PrintTask(name="task3", inputs=[t2])

queue = TaskQueue()
queue.run([t3])
```

### Fan-in, fan-out

```python
from pathlib import Path
from modak import Task, TaskQueue

class DummyTask(Task):
    def run(self):
        self.log_info(f"Running {self.name}")
        for output in self.outputs:
            output.write_text(f"Output of {self.name}")

# Leaf tasks
a = DummyTask(name="A", outputs=[Path("a.out")])
b = DummyTask(name="B", outputs=[Path("b.out")])
c = DummyTask(name="C", outputs=[Path("c.out")])

# Fan-in: D depends on A, B, C
d = DummyTask(name="D", inputs=[a, b, c], outputs=[Path("d.out")])

# Fan-out: E and F both depend on D
e = DummyTask(name="E", inputs=[d], outputs=[Path("e.out")])
f = DummyTask(name="F", inputs=[d], outputs=[Path("f.out")])

queue = TaskQueue()
queue.run([e, f])

```

### A complex workflow

```python
from pathlib import Path
from modak import Task, TaskQueue

class SimTask(Task):
    def run(self):
        self.log_info(f"{self.name} starting with {self.requirements}")
        for out in self.outputs:
            out.write_text(f"Generated by {self.name}")

# Raw data preprocessing
pre_a = SimTask(name="PreA", outputs=[Path("a.pre")], requirements={"cpu": 1})
pre_b = SimTask(name="PreB", outputs=[Path("b.pre")], requirements={"cpu": 1})
pre_c = SimTask(name="PreC", outputs=[Path("c.pre")], requirements={"cpu": 1})

# Feature extraction (can run in parallel)
feat1 = SimTask(name="Feature1", inputs=[pre_a], outputs=[Path("a.feat")], requirements={"cpu": 2})
feat2 = SimTask(name="Feature2", inputs=[pre_b], outputs=[Path("b.feat")], requirements={"cpu": 2})
feat3 = SimTask(name="Feature3", inputs=[pre_c], outputs=[Path("c.feat")], requirements={"cpu": 2})

# Aggregation step
aggregate = SimTask(
    name="Aggregate",
    inputs=[feat1, feat2, feat3],
    outputs=[Path("agg.out")],
    requirements={"cpu": 3}
)

# Final model training (expensive, must be isolated)
train = SimTask(
    name="TrainModel",
    inputs=[aggregate],
    outputs=[Path("model.bin")],
    isolated=True,
    requirements={"cpu": 3, "gpu": 1}
)

# Side analysis and visualization can run independently
viz = SimTask(name="Visualization", inputs=[feat1, feat2], outputs=[Path("viz.png")], requirements={"cpu": 1})
stats = SimTask(name="Stats", inputs=[feat3], outputs=[Path("stats.txt")], requirements={"cpu": 1})

queue = TaskQueue(
    workers=4,
    resources={"cpu": 4, "gpu": 1}
)

queue.run([train, viz, stats])

```

## Future Plans

I'll probably make small improvements to the TUI and add features as I find the
need. Contributions are welcome, just open an issue or pull request on GitHub
and I'll try to respond as soon as I can.
