semvec/pypi-packages huggingface dataset:
motia 1.0.0rc22
Motia framework for III Engine
Requires Python: >=3.10

# Motia Framework for Python

High-level framework for building workflows with the III Engine.

## Installation

```bash
uv pip install motia
```

## Usage

### Defining a Step

```python
from motia import FlowContext, queue

config = {
    "name": "process-data",
    "triggers": [queue("data.created")],
    "enqueues": ["data.processed"],
}

async def handler(data: dict, ctx: FlowContext) -> None:
    ctx.logger.info("Processing data", data)
    await ctx.enqueue({"topic": "data.processed", "data": data})
```

### API Steps

```python
from motia import ApiRequest, ApiResponse, FlowContext, http

config = {
    "name": "create-item",
    "triggers": [http("POST", "/items")],
    "emits": ["item.created"],
}

async def handler(req: ApiRequest, ctx: FlowContext) -> ApiResponse:
    ctx.logger.info("Creating item", req.body)
    await ctx.enqueue({"topic": "item.created", "data": req.body})
    return ApiResponse(status=201, body={"id": "123"})
```

### Streams

```python
from motia import Stream

# Define a stream
todo_stream = Stream[dict]("todos")

# Use the stream
item = await todo_stream.get("group-1", "item-1")
await todo_stream.set("group-1", "item-1", {"title": "Buy milk"})
await todo_stream.delete("group-1", "item-1")
items = await todo_stream.get_group("group-1")
```


### Build & Publish
```bash
python -m build
uv publish --index cloudsmith dist/*
```


## Features

- Event-driven step definitions
- API route handlers
- Cron job support
- Stream-based state management
- Type-safe context with logging

## Testing

### Running Integration Tests

Integration tests require a running III Engine instance. Make sure to have it built or installed before running tests.

1. Install dev dependencies:
   ```bash
   cd motia && uv sync --all-extras

   ```

   ```
3. Run tests:
   ```bash
   uv run pytest
   ```

### Test Configuration

Tests use non-default ports to avoid conflicts:
- Engine WebSocket: `ws://localhost:49199`
- HTTP API: `http://localhost:3199`

Set `III_ENGINE_PATH` environment variable to point to the III engine binary.

### Test Coverage

The integration test suite covers:
- Bridge connection and function registration
- API triggers (HTTP endpoints)
- KV Server operations
- PubSub messaging
- Logging module
- Motia framework integration
- Stream operations (when available)
- State management (when available)
antioch-py 3.1.2
Antioch Python Module SDK
Requires Python: <3.13,>=3.10
Keywords: robotics, simulation, middleware, sdk, modules

# antioch-py

Python SDK for the [Antioch](https://antioch.com) autonomy simulation platform.

## Overview

The antioch-py package provides two components:

### Module SDK (`antioch.module`)

The Module SDK is a framework for building Antioch modules in Python. Modules are containerized components that run alongside your simulation, processing sensor data and producing outputs. Each module runs in its own Docker container and communicates with the simulation through the Antioch runtime. Install the SDK in your module's Dockerfile to read sensors, run inference, and publish results.

```python
from antioch.module import Execution, Module

def process_radar(execution: Execution) -> None:
    scan = execution.read_radar("sensor")
    if scan is not None and len(scan.detections) > 0:
        execution.output("detections").set(scan)

if __name__ == "__main__":
    module = Module()
    module.register("radar_node", process_radar)
    module.spin()
```

### Session SDK (`antioch.session`)

The Session SDK is a client library for orchestrating Antioch simulations. Use it from Python scripts or Jupyter notebooks to programmatically build scenes, load assets, spawn robots, control simulation playback, and record data. The Session SDK connects to your Antioch deployment and provides a high-level API for automation and experimentation.

```python
from antioch.session import Scene, Session, Task, TaskOutcome

session = Session()
scene = Scene()

# Load environment and robot
scene.add_asset(path="/World/environment", name="warehouse", version="1.0.0")
ark = scene.add_ark(name="my_robot", version="0.1.0")

# Run simulation
task = Task()
task.start(mcap_path="/tmp/recording.mcap")

scene.step(1_000_000)  # step 1 second

task.finish(outcome=TaskOutcome.SUCCESS)
```

## Installation

To install in your Python environment:

```bash
pip install antioch-py
```

To install in your Python-based Docker image (e.g. for an Antioch module):

```dockerfile
FROM python:3.12-slim

RUN pip install antioch-py

COPY . /app
WORKDIR /app

CMD ["python", "module.py"]
```

## Documentation

Visit [antioch.com](https://antioch.com) for full documentation.

## License

MIT