Metadata-Version: 2.4
Name: dt-forge
Version: 0.6.0
Summary: Domain-agnostic Python framework for digital twin architectures
License: MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: paho-mqtt>=2.0
Requires-Dist: influxdb-client>=1.40
Requires-Dist: pymongo>=4.6
Requires-Dist: redis>=5.0
Requires-Dist: minio>=7.2
Requires-Dist: asyncpg>=0.29
Requires-Dist: pydantic>=2.5
Requires-Dist: pydantic-settings>=2.1
Requires-Dist: python-dotenv>=1.0
Requires-Dist: fastapi>=0.109
Requires-Dist: uvicorn[standard]>=0.27
Requires-Dist: httpx>=0.26
Requires-Dist: scipy>=1.12
Requires-Dist: numpy>=1.26
Requires-Dist: simpy>=4.1
Requires-Dist: onnxruntime>=1.17
Requires-Dist: scikit-learn>=1.4
Requires-Dist: prophet>=1.1
Requires-Dist: transitions>=0.9
Requires-Dist: simple-pid>=2.0
Requires-Dist: neo4j>=5.17
Requires-Dist: vaderSentiment>=3.3
Requires-Dist: langchain>=0.3
Requires-Dist: langchain-classic>=0.1
Requires-Dist: langchain-community>=0.3
Requires-Dist: langchain-openai>=0.2
Requires-Dist: langchain-anthropic>=0.2
Requires-Dist: langchain-ollama>=0.2
Requires-Dist: stable-baselines3<2.3,>=2.2
Requires-Dist: gymnasium<0.30,>=0.29
Requires-Dist: imitation<2.0,>=1.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: click>=8.1
Requires-Dist: pyserial>=3.5
Provides-Extra: dev
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-cov>=4.1; extra == "dev"
Requires-Dist: ruff>=0.2; extra == "dev"
Requires-Dist: mypy>=1.8; extra == "dev"

# DT-Forge

DT-Forge is a domain-agnostic Python framework for building **digital twins** —
live software models of a real-world asset that ingest its sensor data, store and
model it, react to it, reason about it, and act on it. The same framework works
for a pump, a patch of soil, a plant, an HVAC unit, or a manufacturing line,
because nothing in the core knows anything about your domain. You declare what
your asset's sensors look like, write a small amount of asset-specific glue, and
the framework supplies the rest.

It does that by organising every twin into **six layers** stacked from raw data
at the bottom to autonomous decision-making at the top, plus two cross-cutting
subsystems: **connectors** that let twins talk to each other, and **collection
twins** that group many twins into one. Every layer is optional — you assemble
only the ones your asset needs.

## The six-layer architecture

```
┌─────────────────────────────────────────────────────────────┐
│                 AUTONOMOUS LAYER (Layer 6)                   │
│  OODA loop · GoalPlanner · AutonomousOverseer (LLM)          │
│  RL policy (SAC/TD3/PPO/A2C) · PolicyDeployer · HumanNotifier│
├─────────────────────────────────────────────────────────────┤
│                INTELLIGENT LAYER (Layer 5)                   │
│  MultiAgentSystem · LangChain DiagnosticAgent                │
│  Neo4j KnowledgeGraph · per-agent status & history           │
├─────────────────────────────────────────────────────────────┤
│                 REACTIVE LAYER (Layer 4)                     │
│  ThresholdRuleEngine · MultiStateFSMRuleEngine               │
│  simple-pid PIDController · PostgreSQL RuleRepository         │
├─────────────────────────────────────────────────────────────┤
│                 SERVICES LAYER (Layer 3)                     │
│  Eclipse Ditto sync · FastAPI REST + SSE chat                │
├─────────────────────────────────────────────────────────────┤
│            SIMULATION & MODEL LAYER (Layer 2)                │
│  scipy ODE · ONNX/sklearn surrogate · SimPy · Prophet        │
├─────────────────────────────────────────────────────────────┤
│                    DATA LAYER (Layer 1)                      │
│  MQTT ingest · TelemetryRouter · DataManagementPipeline      │
│  InfluxDB · MongoDB · Redis · MinIO                          │
│  ProvenanceLog · SessionStore · TextIngestor                 │
└─────────────────────────────────────────────────────────────┘

  Cross-cutting:  Connectors        (MQTT · Ditto · HTTP API)
                  Collection twins   (Aggregate · Collection ·
                                      Composite · Network)
  Feeding Layer 6: Learning toolkit  (imitation + inverse RL → policies/rewards)
```

What each layer does:

| Layer | Responsibility | Core building blocks |
|-------|----------------|----------------------|
| **1 — Data** | Ingest telemetry off MQTT, route it to the right stores, smooth it, score asset health, keep an audit trail | `MQTTIngestor`, `TelemetryRouter`, `DataManagementPipeline`, the `Influx`/`Mongo`/`Redis`/`MinIO` adapters, `ProvenanceLog`, `SessionStore`, `TextIngestor` |
| **2 — Simulation** | Predict, forecast and detect drift against a model of the asset | `ODEModel` (scipy), `ONNXSurrogate` / `SKLearnSurrogate`, `SimPyModel`, `ProphetForecaster`, `ModelRunner` |
| **3 — Services** | Expose the twin: sync canonical state to Eclipse Ditto, serve a REST + SSE API | `DittoSyncService`, the FastAPI app, the SSE chat stream |
| **4 — Reactive** | Turn readings into actions automatically: thresholds, a state machine, closed-loop control | `ThresholdRuleEngine`, `MultiStateFSMRuleEngine`, `PIDController`, `RuleRepository` |
| **5 — Intelligent** | Diagnose *why* something is wrong using a knowledge graph and LLM agents | `KnowledgeGraph` (Neo4j), `DiagnosticAgent` (LangChain), `MultiAgentSystem` |
| **6 — Autonomous** | Decide *what to do next*: observe-orient-decide-act, goal planning, learned RL policies, human escalation | `OODALoop`, `GoalPlanner`, `AutonomousOverseer`, `PolicyTrainer`/`PolicyDeployer`, `HumanNotifier` |

A monitoring-only twin needs just the Data layer (plus MQTT ingest). A
self-managing asset stacks all six. You never edit the framework to choose — you
simply return the layers you want from one method (`build_layers`).

## Collection twins

Real systems are rarely a single asset. Four collection patterns let you treat a
group of twins as one twin, each suited to a different topology:

| Pattern        | Use it for                                                        |
|----------------|-------------------------------------------------------------------|
| `AggregateDT`  | A fleet of *identical* assets — fused state and shared control     |
| `CollectionDT` | Batch monitoring, outlier detection, statistical comparison        |
| `CompositeDT`  | Hierarchical systems that exchange boundary conditions, with swap  |
| `NetworkDT`    | Graph-topology systems — cascade risk and bottleneck detection     |

## Connectors

Connectors are the cross-twin transport. A twin can publish to and subscribe
from another twin over `MQTTConnector`, `DittoConnector`, or `APIConnector`
(HTTP), all managed through a `ConnectorRegistry`. This is what makes
*cross-domain* twins possible: a soil twin can feed a plant twin, which can feed
a yield-forecasting twin.

## Installation

DT-Forge requires **Python 3.11+**. The backing services (MQTT broker,
databases, Ditto, Neo4j) run in Docker, so you also need Docker with the Compose
plugin.

```bash
pip install dt-forge
```

That installs the framework and the `dtforge` command-line tool. Current version:
**0.4.3**.

## Quick start

### 1. Configure

All configuration comes from environment variables with the `DT_` prefix.
Nested fields use a **double-underscore** delimiter — `DT_MQTT__BROKER`, not
`DT_MQTT_BROKER` (the single-underscore form is silently ignored). Put them in a
`.env` file in your project directory:

```bash
DT_ASSET_ID=pump_001
DT_ASSET_TYPE=centrifugal_pump
DT_ASSET_NAME="Plant A Pump"

DT_MQTT__BROKER=localhost
DT_MQTT__PORT=1883

DT_INFLUX__URL=http://localhost:8086
DT_INFLUX__TOKEN=my-super-secret-token
DT_INFLUX__ORG=digital_twin
DT_INFLUX__BUCKET=asset_telemetry

DT_MONGO__URI=mongodb://admin:password@localhost:27017
DT_REDIS__URL=redis://localhost:6379
DT_DITTO__URL=http://localhost:8080

# Optional — only for the intelligent layer
DT_LLM__PROVIDER=anthropic        # openai | anthropic | ollama
DT_LLM__MODEL=claude-sonnet-4-6
DT_LLM__API_KEY=sk-ant-...
DT_NEO4J__URI=bolt://localhost:7687
```

Every field has a sensible default (see `dt_forge/core/config.py`), so you only
set what differs from the defaults.

### 2. Start the infrastructure you need

`dtforge infra up` writes a `docker-compose.yml` containing exactly the services
your chosen layers require, then runs `docker compose up -d`:

```bash
# Minimal monitoring twin (MQTT broker + the four data stores):
dtforge infra up --layers data,network

# Add Eclipse Ditto (services) and the Neo4j knowledge graph (intelligent):
dtforge infra up --layers data,network,services,intelligent

# Write the compose file without starting anything:
dtforge infra up --layers data,network,services --generate-only
docker compose up -d
```

Each layer maps to a fixed set of containers:

| Service        | Host port(s) | Provisioned for the layer(s) | Purpose                          |
|----------------|--------------|------------------------------|----------------------------------|
| Mosquitto      | 1883, 9001   | `network`                    | MQTT broker (+ WebSocket)        |
| InfluxDB       | 8086         | `data` or `network`          | Time-series telemetry            |
| MongoDB        | 27017        | `data` or `network`          | Events + provenance              |
| Redis          | 6379         | `data` or `network`          | Cache, FSM state, sessions       |
| MinIO          | 9000         | `data` or `network`          | Trained models, large objects    |
| Eclipse Ditto  | 8080         | `services`                   | Canonical twin state (+ nginx)   |
| Neo4j          | 7474, 7687   | `intelligent`                | Knowledge graph                  |
| Grafana        | 3000         | always                       | Dashboards / observability       |

> **PostgreSQL is not provisioned by `infra up`.** The optional
> `RuleRepository` (versioned, hot-reloadable reactive rules) talks to Postgres
> through `asyncpg`; if you use it, point it at a Postgres instance you run
> yourself. Everything else in the reactive layer works without it.

The command also records your chosen layers in a `.dtforge-layers` file. Once the
containers are up, verify the twin can reach each one:

```bash
dtforge infra check          # reads .dtforge-layers automatically
```

Every reachable service prints a `✓`.

### 3. Scaffold a twin

```bash
dtforge init \
  --asset-type centrifugal_pump \
  --name "Plant A Pump" \
  --asset-id pump_001
```

That writes two files into the output directory:

| File      | Purpose                                                        |
|-----------|----------------------------------------------------------------|
| `twin.py` | A runnable twin class — edit `build_layers()` to wire your asset|
| `.env`    | Environment configuration, pre-filled with the defaults        |

The generated `twin.py` already wires the Data, MQTT-ingest, data-management,
Ditto-sync and reactive layers, so it runs as soon as you fill in your sensor
fields. Add the simulation, intelligent and autonomous layers as your twin grows.

### 4. Run

```bash
python twin.py            # run the module directly
# or, from the project directory:
dtforge run twin          # imports twin.py and drives its lifecycle
```

If you wired the services layer, the FastAPI app serves these endpoints:

| Endpoint                     | Returns                                            |
|------------------------------|----------------------------------------------------|
| `GET  /health`               | Liveness — `{asset_id, status}`                    |
| `GET  /api/twin/state`       | Canonical twin state from Ditto                    |
| `GET  /api/twin/telemetry`   | Latest telemetry feature from Ditto                |
| `GET  /api/twin/health-score`| Current health-score feature                       |
| `GET  /api/twin/events`      | Recent events (needs a doc store)                  |
| `POST /api/twin/external`    | Inbound push from another twin's `APIConnector`    |
| `POST /api/chat`             | SSE stream of the LLM diagnostic agent's replies   |

## A monitoring twin in ~30 lines

```python
import asyncio, logging
from dotenv import load_dotenv
from dt_forge.core.config import TwinConfig, SensorFieldSpec
from dt_forge.core.base import AbstractDigitalTwin
from dt_forge.core.lifecycle import TwinLifecycle
from dt_forge.data import InfluxAdapter, MongoAdapter, RedisAdapter
from dt_forge.data.writer import TelemetryRouter
from dt_forge.data.management import DataManagementPipeline
from dt_forge.network import MQTTIngestor
from dt_forge.reactive import ThresholdRuleEngine

load_dotenv()
logging.basicConfig(level=logging.INFO)

config = TwinConfig(sensor_fields=[
    SensorFieldSpec(name="temperature_c", nominal=25.0, noise_std=0.5,
                    warn_threshold=60.0, crit_threshold=75.0),
    SensorFieldSpec(name="pressure_bar", nominal=4.0, noise_std=0.05,
                    warn_threshold=3.0, crit_threshold=2.0,
                    threshold_direction="low"),
])

class PumpTwin(AbstractDigitalTwin):
    def build_layers(self):
        ts, doc, cache = InfluxAdapter(self.config), MongoAdapter(self.config), RedisAdapter(self.config)
        router = TelemetryRouter(self.config, self.bus,
                                 ts_store=ts, doc_store=doc, cache=cache)
        return {
            "data":      router,
            "network":   MQTTIngestor(self.config, self.bus, router=router),
            "data_mgmt": DataManagementPipeline(self.config, self.bus,
                                                ts_store=ts, cache=cache),
            "reactive":  ThresholdRuleEngine(self.config, self.bus,
                                             ts_store=ts, cache=cache, doc_store=doc),
        }

if __name__ == "__main__":
    lc = TwinLifecycle(); lc.add(PumpTwin(config))
    asyncio.run(lc.run_forever())
```

Point a real device (or `dt_forge.physical.simulator.GenericSimulator`) at the
topic `dt/pump_001/telemetry`, and the twin will route every reading to InfluxDB
and MongoDB, smooth it, score health, and raise warning/critical events through
the threshold engine — all automatically. Layer a simulation model, a knowledge
graph, an LLM agent and an OODA loop on top of this skeleton when you need them.

## Teaching a twin from demonstrations

The autonomous layer's RL policies need a reward function. When the behaviour you
want is a human skill you can't easily write down as a reward, the
`dt_forge.learning` toolkit lets you learn it from examples instead:

- **Imitation** — clone an expert's actions directly (`BCTrainer`, `DAggerTrainer`).
- **Inverse RL** — *recover the reward function* behind an expert's behaviour
  (`AIRLTrainer`, `GAILTrainer`, `MaxEntIRLTrainer`).
- **Plumbing** — `FeatureSpec` (one source of truth for the observation vector),
  `Demonstrations`/`DemonstrationSource`, `LearnedRewardFn` (reuse a recovered
  reward in ordinary RL), and `SkillTransferPipeline` (chain BC → IRL → RL, then
  validate, version and promote).

A recovered reward plugs straight back into the same `GenericTwinEnv` and
`PolicyDeployer` the autonomous layer already uses.

## Extension points

The framework is built around protocols and abstract base classes — swap any
piece without touching the rest:

| Extend                  | By                                                    |
|-------------------------|-------------------------------------------------------|
| Storage backend         | Implementing the `TimeSeriesStore` / `DocumentStore` /|
|                         | `CacheStore` / `ObjectStore` protocol                 |
| Physics model           | Subclassing `ODEModel` or implementing `TwinModel`    |
| ML surrogate            | Implementing `TwinModel` over ONNX or scikit-learn    |
| Discrete-event model    | Wrapping a generator with `SimPyModel`                |
| Forecaster              | Wrapping a time-series model with `ProphetForecaster` |
| Custom reactive rule    | Implementing `Rule.evaluate(readings)`                |
| N-state FSM             | Subclassing `MultiStateFSMRuleEngine`                 |
| Diagnostic agent tools  | Overriding `DiagnosticAgent._build_extra_tools()`     |
| LLM provider            | Setting `DT_LLM__PROVIDER` (openai/anthropic/ollama)  |
| Goal-based assessment   | Subclassing `GoalPlanner.assess()`                    |
| Strategic LLM decisions | Wiring an `AutonomousOverseer` into the `OODALoop`    |
| RL policy               | `PolicyTrainer` + `PolicyDeployer` (Stable-Baselines3)|
| Reward function         | Passing `reward_fn` to `GenericTwinEnv`, or learning  |
|                         | one with `dt_forge.learning`                          |
| Notification channel    | Implementing a `NotificationBackend` (email/Slack/web)|
| Cross-twin transport    | Implementing `ConnectorProtocol`                      |
| New collection pattern  | Subclassing `AbstractCollectionTwin`                  |
| Domain session state    | Subclassing `SessionContext`, using `SessionStore[T]` |

## Technology stack

| Concern          | Library                                            |
|------------------|----------------------------------------------------|
| Messaging        | Eclipse Mosquitto / paho-mqtt 2.x                  |
| Time-series      | InfluxDB 2 (`influxdb-client`)                     |
| Documents        | MongoDB (`pymongo` + `motor`)                      |
| Cache / pub-sub  | Redis                                              |
| Object storage   | MinIO (S3 API)                                     |
| Relational       | PostgreSQL (`asyncpg`) — optional, for `RuleRepository` |
| Twin state       | Eclipse Ditto                                      |
| Knowledge graph  | Neo4j 5                                            |
| Web API          | FastAPI + Uvicorn + sse-starlette                  |
| Config           | Pydantic v2 + pydantic-settings                    |
| State machine    | `transitions`                                      |
| PID control      | `simple-pid`                                       |
| Physics          | SciPy (`solve_ivp`)                                |
| Surrogate / ML   | ONNX Runtime + scikit-learn                        |
| Forecasting      | Prophet                                            |
| Discrete event   | SimPy                                              |
| Sentiment / NLP  | vaderSentiment (swappable for any callable)        |
| Agents           | LangChain (+ langchain-classic / -community)       |
| LLM providers    | OpenAI · Anthropic · Ollama                        |
| RL               | Stable-Baselines3 + Gymnasium                      |
| Learning-from-demo | `imitation` (BC, DAgger, AIRL, GAIL) + a built-in MaxEnt IRL |
| CLI              | Click                                              |

## CLI reference

```bash
dtforge init        --asset-type TYPE --name NAME --asset-id ID [--out DIR]
dtforge infra up    --layers data,network,services[,intelligent] [--out FILE] [--generate-only]
dtforge infra check [--layers ...]              # default: read .dtforge-layers
dtforge run         [TWIN_MODULE]               # default module: "twin"
dtforge train       [--algorithm SAC|TD3|PPO|A2C] [--timesteps N]
                    [--env-module M] [--save NAME]
```

Run any command with `--help` for the full flag list.

## Learn more

The `guide/` folder in the source repository is the complete developer manual —
twelve chapters that take you from the mental model to a full worked example,
layer by layer.

## License

MIT
