Development

This page summarizes the local development workflow for MicroDCS, including environment setup, test execution, documentation work, and container packaging.

Environment Setup

MicroDCS requires Python ≥ 3.14 and uses uv for dependency management, virtual environment creation, and packaging.

# Install dependencies and create the local virtual environment
uv sync

Why Python 3.14?

Python 3.14 is the minimum version where the annotation semantics MicroDCS needs work correctly end-to-end without workarounds. The requirement is driven by PEP 749 (Deferred Evaluation of Annotations) and the new annotationlib module.

MicroDCS relies heavily on runtime type inspection:

• mashumaro resolves type annotations at runtime to build its JSON (orjson) and MessagePack serialization codecs. This means annotations must be real types at runtime, not opaque strings.
• DataClassResponseMixin[R] uses Generic[R] with runtime introspection of __orig_bases__ to determine the response type. Generated model code like DataClassResponseMixin["Hello"] passes a forward reference that must resolve correctly.
• ProtocolBinding[PH] resolves its generic type parameter at runtime via get_args() to find the associated ProtocolHandler class, including lazy ForwardRef resolution through the 3.14 annotationlib.ForwardRef.
• Generated dataclasses use InitVar[Hello | None], union type aliases (type Greetings = Hello | Bye), and forward references freely. These all need to round-trip through typing.get_type_hints() correctly.
• register_callback() unwraps TypeAliasType (the runtime object behind the type X = Y statement) and UnionType (X | Y) to register individual dataclass handlers.

Before Python 3.14, there were two problematic alternatives:

1. PEP 563 (from __future__ import annotations) — stringifies all annotations at definition time. This breaks mashumaro's runtime codegen, get_type_hints() resolution of generic parameters, and any code that inspects annotations as live type objects.
2. No PEP 563 on older Python — requires careful class ordering to avoid NameError on forward references, manual update_forward_refs() calls, and if TYPE_CHECKING guards with duplicated imports. This is fragile and error-prone with generated code.

PEP 749 solves this by making annotation evaluation lazy by default: annotations are stored in a form that is evaluated on access rather than at class creation time. This means forward references, union syntax, and type aliases all work correctly both as static type information and when inspected at runtime — exactly the semantics mashumaro, get_type_hints(), and the generic resolution patterns in MicroDCS require.

Additional 3.14 features used throughout the codebase:

• annotationlib.ForwardRef — the new lazy forward reference type, used in ProtocolBinding.get_protocol_handler() for generic parameter resolution
• type statement (PEP 695, Python 3.12+) — used for type aliases like type Greetings = Hello | Bye in generated models and type JobOrderIdCall = AbortCall | CancelCall | ... in processors
• StrEnum (Python 3.11+) — used for Direction, MessageIntent, ProcessorBinding, and ErrorKind enums
• kw_only dataclasses (Python 3.10+) — used universally for all model dataclasses
• match/case (Python 3.10+) — used in validation and content-type routing

Common Commands

Run the main unit test suite:

uv run pytest tests/unit

Run test coverage:

uv run pytest --cov=microdcs --cov-report=term-missing tests/unit

Run benchmarks and save results:

uv run pytest --benchmark-only --benchmark-autosave

Compare against a previously saved baseline:

# Save current results as the named baseline
uv run pytest --benchmark-only --benchmark-save=baseline

# Later, compare against that baseline
uv run pytest --benchmark-only --benchmark-compare=baseline

Benchmarks are automatically disabled during normal test runs. They only execute when --benchmark-only is passed, or when running the full suite with --benchmark-autosave (which runs them alongside regular tests and persists results under .benchmarks/).

The benchmarked tests cover the per-message hot paths:

• test_common.py — CloudEvent JSON and msgpack round-trips; payload serialization/deserialization
• test_dataclass.py — DataClassMixin JSON and msgpack round-trips
• test_sfc_recipe.py — SfcStep and full SfcRecipe JSON and msgpack round-trips
• statemachine_test.py — ISA95 job state machine trigger dispatch

Run the example application:

# Requires a reachable MQTT broker and Redis server
uv run python -m app

MessagePack Buffer Limit

The MessagePack-RPC transport sets msgpack.Unpacker(max_buffer_size=...) to bound the size of buffered, not-yet-decoded payload data.

• Default limit: 8388608 bytes (8 MiB)
• Runtime override: set APP_MSGPACK_MAX_BUFFER_SIZE

This helps protect the process from excessive memory usage caused by oversized or malformed input streams.

Configuration Validation

At startup, the runtime configuration is validated before protocol handlers enter their main loops. Validation fails fast with a clear error message if any check does not pass.

Checked constraints include:

• Required non-empty values:
  • instance_id
  • redis.hostname
  • redis.key_prefix
  • mqtt.hostname
  • mqtt.identifier
  • msgpack.hostname
• Files that must exist when TLS is required:
  • mqtt.tls_cert_path (with mqtt.require_tls)
  • msgpack.tls_server_cert_path and tls_server_key_path (with msgpack.require_tls or tls_client_auth)
  • msgpack.tls_cert_path (with msgpack.tls_client_auth)
  • mqtt.tls_client_cert_path and tls_client_key_path (when set; both are required together)
• Port ranges:
  • redis.port, mqtt.port, msgpack.port must be in 1..65535
• Positive values:
  • mqtt.connect_timeout, mqtt.publish_timeout, mqtt.message_workers
  • mqtt.dedupe_lease_seconds
  • redis.socket_timeout, redis.socket_connect_timeout
  • msgpack.max_queued_connections, msgpack.max_concurrent_requests
  • msgpack.max_buffer_size
• Non-negative values:
  • mqtt.incoming_queue_size, mqtt.outgoing_queue_size
  • mqtt.dedupe_ttl_seconds, mqtt.binding_outgoing_queue_size
  • msgpack.binding_outgoing_queue_size
  • processing.shutdown_grace_period
  • processing.message_expiry_interval (when configured)

Startup pre-check sequence:

1. Validate static field constraints (required fields, ranges, and bounds).
2. Check TCP connectivity to Redis using redis.hostname:redis.port.
3. Check TCP connectivity to MQTT broker using mqtt.hostname:mqtt.port.
4. Check MessagePack server bindability on msgpack.hostname:msgpack.port.

Environment variables are still parsed from APP_* settings, then validated by these runtime checks.

Deployment Modes

A MicroDCS application consists of two roles: processor (handles incoming events, runs state machines) and publisher (maintains retained MQTT topics for northbound MES integration). Both can run in a single process (default) or be split into separate Kubernetes deployments.

Instance Role Flags

For production, set one flag to false per deployment: multiple processor replicas with shared MQTT subscriptions and a single publisher replica that owns retained topics.

Publisher Configuration (APP_PUBLISHER_*)

The topic prefix identifier (e.g. "machinery-jobs") is a constructor parameter of JobOrderPublisher, not an environment variable. It is wired in app/__main__.py.

Kubernetes Manifest

The deploy/k8s.yaml manifest defines two Deployments:

• microdcs-processor (replicas: 2) — processors plus SFC engine (APP_IS_PUBLISHER_INSTANCE=false)
• microdcs-publisher (replicas: 1) — publisher only (APP_IS_PROCESSOR_INSTANCE=false)

The publisher must be a single replica to avoid conflicting retained writes. The SFC engine is different: it runs on every processor replica and relies on Redis consumer groups plus CAS Lua scripts for work distribution and recovery. Equipment integrations must treat push_command deliveries as idempotent by deduplicating on the mdcsactionkey attached by the engine (see the idempotency contract). The app reconnects to MQTT with backoff but does not reconnect to Redis: a lost Redis connection ends the process and Kubernetes restarts it, so health probes on external dependencies are not included. See Redis Requirements for the persistence baseline.

SFC Engine Wiring

The relevant SFC wiring in app/__main__.py is:

from microdcs.sfc_engine import SfcEngine

machinery_jobs_processor = MachineryJobsCloudEventProcessor(
    microdcs.runtime_config.instance_id,
    microdcs.runtime_config.processing,
    "machinery-jobs",
    microdcs.redis_connection_pool,
    microdcs.redis_key_schema,
    job_acceptance_config,
)

sfc_engine = SfcEngine(
    microdcs.redis_connection_pool,
    microdcs.redis_key_schema,
    nb_processor=machinery_jobs_processor,
    sb_processors={"greetings": greetings_processor},
    consumer_name=microdcs.runtime_config.instance_id,
)

greetings_processor.register_action_completion_handler(sfc_engine.complete_action)
greetings_processor.register_action_failure_handler(sfc_engine.fail_action)
# For pull_event actions also add:
# greetings_processor.register_pull_completion_handler(sfc_engine.pull_event_handler)
machinery_jobs_processor.register_scope_handler(sfc_engine.register_scope)
microdcs.add_additional_task(sfc_engine)

The register_pull_completion_handler line is required when the recipe uses pull_event actions — it registers sfc_engine.pull_event_handler, which writes the incoming CloudEvent as a pull_event: work item into sfc:work:{scope} so any live engine replica can complete the waiting action. The example recipe uses only push_command, so this wiring is omitted in app/__main__.py.

tests/example_sfc.py provides a reusable build_example_work_master() helper containing an SFC recipe payload so you can seed Redis or publish a ConfigWorkMaster event during manual testing.

Local Development

For local development, both roles run in a single process (default). Start MQTT and Redis via Docker and run the app:

docker start mosquitto 2>/dev/null || docker run --rm -it --name mosquitto -p 1883:1883 eclipse-mosquitto mosquitto -c /mosquitto-no-auth.conf
docker start redis 2>/dev/null || docker run --rm -it --name redis -p 6379:6379 redis:latest
uv run python -m app

Or use the VS Code task Run App (plain) which starts both services automatically.

A compose.yaml in the repository root starts the same infrastructure plus the Aspire dashboard (unauthenticated, local use only):

docker compose up -d                          # Mosquitto, Redis, Aspire dashboard
docker compose --profile app up -d --build    # additionally build and run the example app

For manual SFC testing, use the example Work Master from tests/example_sfc.py and publish a StoreAndStartCall that references it.

Generate typed models from a JSON Schema file:

uv run microdcs dataclassgen dataclasses my-schema.schema.json

Generate models with advanced options (custom metadata, init fields, validation, and root-union workaround):

uv run microdcs dataclassgen dataclasses \
    --custom-metadata \
    --init-fields 'mystatus->MyStatus' \
    --validation \
    --collapse-root-workaround \
    my-schema.schema.json

Generate the SFC recipe dataclasses:

uv run microdcs dataclassgen dataclasses sfc_recipe.schema.json

OpenTelemetry Analysis with Aspire Dashboard

The Aspire dashboard can be run as a standalone container to visualise traces, metrics, and structured logs from any OpenTelemetry-enabled app — no .NET or full Aspire installation required.

Start the dashboard container:

docker start aspire-dashboard 2>/dev/null || docker run --rm -p 18888:18888 -p 4317:18889 -p 4318:18890 -d --name aspire-dashboard mcr.microsoft.com/dotnet/aspire-dashboard:latest

Port mapping:

Open http://localhost:18888 in the browser. The dashboard is secured with a login token by default — retrieve it from the container logs:

docker logs aspire-dashboard 2>&1 | grep "login?t="

To run the app and send telemetry to the dashboard, configure the OTLP exporter and start via opentelemetry-instrument:

APP_PROCESSING_OTEL_INSTRUMENTATION_ENABLED=true \
OTEL_SERVICE_NAME=microdcs.app \
OTEL_TRACES_EXPORTER=otlp \
OTEL_METRICS_EXPORTER=otlp \
OTEL_LOGS_EXPORTER=otlp \
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
OTEL_EXPORTER_OTLP_PROTOCOL=grpc \
OTEL_PYTHON_LOG_CORRELATION=true \
OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED=true \
  .venv/bin/opentelemetry-instrument python -m app

Or use the VS Code task Run App (instrumented/Aspire dashboard) which starts MQTT, Redis, and the Aspire dashboard automatically.

Documentation

The documentation site is built with Zensical.

# Start the local docs server
zensical serve

Container Build

The project ships with a distroless multi-stage Docker build.

docker build -t aschamberger/microdcs .

Main Libraries

The framework is built around a small set of core dependencies:

• aiomqtt and Paho MQTT Python for MQTT communication
• mashumaro for dataclass serialization
• msgpack-python for MessagePack encoding
• redis-py for persistence
• datamodel-code-generator for model generation
• Typer for CLI tooling

Reference Links

• Crow's Nest MQTT
• Eclipse Mosquitto Docker image
• Astral uv Docker integration guide
• Distroless container images tutorial
• Distroless Python with uv
• Mermaid sequence diagram syntax