Metadata-Version: 2.5
Name: vda5050-sdk
Version: 0.2.0
Summary: Async Python SDK for the VDA 5050 AGV communication standard (master control and vehicle side).
Project-URL: Homepage, https://gitee.com/zwl1336849005/vda5050-sdk
Project-URL: Repository, https://gitee.com/zwl1336849005/vda5050-sdk
Project-URL: Issues, https://gitee.com/zwl1336849005/vda5050-sdk/issues
Project-URL: Changelog, https://gitee.com/zwl1336849005/vda5050-sdk/blob/main/CHANGELOG.md
Author: VDA5050 SDK Contributors
License-Expression: MIT
License-File: LICENSE
Keywords: agv,amr,fleet,logistics,mqtt,robotics,vda5050,wcs
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: jsonschema>=4.20
Requires-Dist: pydantic>=2.5
Provides-Extra: all
Requires-Dist: aiomqtt>=2.0; extra == 'all'
Requires-Dist: mypy>=1.10; extra == 'all'
Requires-Dist: pytest-asyncio>=0.23; extra == 'all'
Requires-Dist: pytest>=8.0; extra == 'all'
Requires-Dist: ruff>=0.5; extra == 'all'
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: mqtt
Requires-Dist: aiomqtt>=2.0; extra == 'mqtt'
Provides-Extra: release
Requires-Dist: build>=1.2; extra == 'release'
Requires-Dist: twine>=5.0; extra == 'release'
Description-Content-Type: text/markdown

# vda5050-sdk

> Async Python SDK for the [VDA 5050](https://github.com/VDA5050/VDA5050) AGV/AMR communication
> standard — usable from **both** the master-control (WCS) side and the vehicle (AGV) side.

[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![Python](https://img.shields.io/badge/python-3.10%2B-blue.svg)](pyproject.toml)
[![Typing: py.typed](https://img.shields.io/badge/typing-py.typed-informational.svg)](src/vda5050/py.typed)

`vda5050-sdk` is a **library**, not a service. It gives you typed protocol models, schema
validation, an order builder with correct `sequenceId` / base-horizon / stitching semantics, an
action lifecycle, and pluggable adapters for MQTT and persistence. Your application keeps its own
process, event loop and business orchestration (job scheduling, PLC interlocks, traffic rules).

---

## Features

- **Full protocol data model** for `order`, `instantActions`, `state`, `connection`,
  `visualization` and `factsheet` (pydantic v2, camelCase on the wire).
- **JSON-Schema validation** against the official VDA 5050 schemas (bundled).
- **Order domain logic**: sequence-id allocation across nodes + edges, base/horizon handling,
  order-update **stitching**, and the 9-step order-acceptance decision.
- **Action model**: blocking types (`NONE` / `SOFT` / `HARD`), lifecycle state machine and timeouts.
- **Arrival verification**: a node is only marked traversed once the vehicle is on the right
  map, with an initialized position, within `allowedDeviationXY/Theta`; failures halt the
  vehicle and report a FATAL error.
- **Order/state reconciliation**: master-side `OrderProgressMonitor` detects wrong orders, stale
  updates, unknown nodes, stuck actions, stalled progress and map mismatches.
- **Closed-loop master events**: `on_vehicle_error`, `on_base_request` (auto `extend_base` via
  `BaseRequestStrategy`), `on_information`, `on_visualization`, plus `supervise()` health checks
  with auto `stateRequest` and offline detection.
- **Master client** (`MasterClient`): dispatch / extend-base / cancel / instant actions.
- **Vehicle client** (`VehicleClient`): receive orders, execute actions, publish state (event-driven
  + max 30 s interval).
- **Multi-vehicle** support: one `VehicleSession` per AGV, per-vehicle state persistence.
- **Pluggable ports**: swap MQTT library, storage or clock without touching domain logic.
- **Optional orchestration**: PLC-gated transport-job strategies.
- **Map orchestration**: `downloadMap` / `enableMap` / `initPosition` helpers plus
  `dispatch(..., require_maps=True)` pre-checks for multi-floor sites.
- **Safety rails**: optional inbound protocol-version and JSON-Schema validation
  (`validate_protocol_version` / `validate_messages`).
- **Testing helpers**: in-memory transport and a fake AGV for fast integration tests.

## Installation

```bash
pip install vda5050-sdk            # core (models, domain, validation)
pip install "vda5050-sdk[mqtt]"    # + async MQTT adapter (aiomqtt)
pip install "vda5050-sdk[dev]"     # + test/lint tooling
```

## Quick start — master control side

```python
import asyncio
from vda5050 import MasterClient, MasterConfig, OrderBuilder, Action, BlockingType

async def main() -> None:
    cfg = MasterConfig(broker_url="mqtt://localhost:1883")
    async with MasterClient(cfg) as mc:
        mc.on_state(lambda s, key: print("state:", key, s.order_id, s.last_node_id))
        mc.on_error(lambda e: print("error:", e.error_type, e.error))
        mc.on_connection(lambda c, key: print("connection:", key, c.connection_state))

        order = (
            OrderBuilder(order_id="ORDER-1", manufacturer="ACME", serial_number="AGV-01")
            .add_node("AISLE_01", x=5.0, y=2.0, theta=0.0, map_id="M1")
            .add_edge("AISLE_01", "STATION_05")
            .add_node(
                "STATION_05", x=10.0, y=2.0, theta=0.0, map_id="M1",
                actions=[Action(action_id="a-pick-1", action_type="pick",
                                blocking_type=BlockingType.HARD,
                                action_parameters=[{"key": "loadType", "value": "EPAL"}])],
            )
            .build()
        )
        await mc.dispatch(order, target="ACME/AGV-01")

asyncio.run(main())
```

## Quick start — vehicle side

```python
import asyncio
from vda5050 import ActionStatus, VehicleClient, VehicleConfig

async def main() -> None:
    cfg = VehicleConfig(manufacturer="ACME", serial_number="AGV-01",
                        broker_url="mqtt://localhost:1883")
    async with VehicleClient(cfg) as vc:
        @vc.on_order
        async def handle_order(order) -> None:
            print("got order", order.order_id)
            for node in order.nodes:
                await vc.update_position(x=node.node_position.x, y=node.node_position.y,
                                         theta=node.node_position.theta,
                                         map_id=node.node_position.map_id)
                for action in node.actions:
                    await vc.report_action(action_id=action.action_id,
                                           status=ActionStatus.FINISHED)

asyncio.run(main())
```

## Project layout

```
src/vda5050/
  model/          # protocol DTOs (pure, no I/O)
  serialization/  # header + JSON codec
  validation/     # JSON-Schema validation (bundled schemas)
  domain/         # order / action / session / map / capability logic
  ports/          # abstract interfaces (transport, store, clock, plc)
  adapters/       # mqtt + persistence implementations
  client/         # MasterClient + VehicleClient facades
  orchestration/  # PLC-gated job strategies (optional)
  testing/        # in-memory transport, fake AGV
```

## Schema sync

The official JSON schemas are vendored under `src/vda5050/validation/schemas/`. To refresh them
from a checkout of the specification repository:

```bash
python scripts/sync_schemas.py --source path/to/VDA5050/json_schemas
```

## Running the tests

```bash
pip install "vda5050-sdk[dev]"
pytest -q
```

## License & attribution

This project is licensed under the [MIT License](LICENSE).

The VDA 5050 protocol specification, its documentation and the JSON schemas are the work of the
VDA 5050 community and are used here for interoperability purposes; see the
[VDA5050 repository](https://github.com/VDA5050/VDA5050) for their terms.
