deta

AI Cost Tracking

Features

Installation

pip install deta

# web dashboard (FastAPI + uvicorn + WebSocket)
pip install 'deta[web]'

# all extras
pip install deta[docker,toml,web]

Usage

Scan infrastructure

deta scan /path/to/project --depth 3 --output infra-map.json

Generate graph outputs (YAML / Mermaid / PNG)

# text graph + mermaid + json
deta scan /path/to/project --formats json,yaml,mermaid

# include online checks (localhost probes)
deta scan /path/to/project --formats json,yaml,mermaid --online

# try PNG (requires graphviz python package + graphviz binary)
deta scan /path/to/project --formats png

Monitor in real-time

deta monitor /path/to/project --interval 30 --depth 3
deta monitor . --interval 30 --depth 3

# realtime watch from scan command (regenerates outputs on each change)
deta scan /path/to/project --watch --formats json,yaml,mermaid --online

Web Dashboard

deta web /path/to/project --depth 3 --host 127.0.0.1 --port 8765

web:
  enabled: true
  host: 127.0.0.1
  port: 8765
  refresh_seconds: 5
  cache_ttl_seconds: 30.0
  debounce_seconds: 0.5
  push_events:
    - service_added
    - service_removed
    - service_up
    - service_down

Compare with baseline

deta diff --baseline infra-map.json /path/to/project

Example Workflow (e.g., for c2004 project)

Python API

from pathlib import Path
from deta import build_topology, scan_compose, scan_openapi, probe_all

# Build topology from manifests
topology = build_topology(Path("/path/to/project"), max_depth=3)

# Detect anomalies
anomalies = topology.detect_anomalies()
for anomaly in anomalies:
    print(f"{anomaly['severity']}: {anomaly['type']}")

# Cycle & hub analysis
cycles = topology.detect_cycles()
hubs = topology.find_hubs(min_degree=3)

# Export to JSON
import json
output = json.loads(topology.to_json())

# Probe services
results = probe_all(topology.services, max_concurrency=20)

Architecture

deta/
├── scanner/          # Manifest parsing
│   ├── compose.py    # docker-compose.yml (with override merging)
│   ├── openapi.py    # OpenAPI specs
│   ├── npm.py        # package.json
│   ├── python.py     # pyproject.toml & requirements.txt
│   ├── env.py        # .env file loading & interpolation
│   └── ports.py      # Port binding parsing & resolution
├── builder/          # Topology construction
│   ├── topology.py   # Graph, anomaly detection, cycles, hubs
│   └── cache.py      # mtime-based topology cache
├── monitor/          # Real-time monitoring
│   ├── watcher.py    # File change polling
│   ├── prober.py     # HTTP/TCP health checks (concurrent)
│   └── alerter.py    # Rich console output
├── formatter/        # Output formats
│   ├── graph.py      # YAML graph, Mermaid, PNG
│   └── toon.py       # Semcod toon format
├── dsl/              # Change tracking DSL
│   └── commands.py   # service_up/down, port_added/removed
├── web/              # Real-time dashboard
│   └── app.py        # FastAPI + WebSocket (topology, probes, events)
├── integration/      # Ecosystem hooks
│   └── semcod.py     # sumd, pyqual, vallm, pre_deploy_check
├── config.py         # deta.yaml configuration (Pydantic models)
├── core.py           # Base Wup class
└── cli.py            # CLI entry point (scan, monitor, diff, web)

Configuration

Semcod Integration

from pathlib import Path
from deta.integration import generate_for_sumd, generate_for_pyqual, generate_for_vallm, pre_deploy_check

# Generate toon for sumd
generate_for_sumd(Path("."), output=Path("infra.toon.yaml"))

# Quality analysis for pyqual
generate_for_pyqual(Path("."), depth=3)

# Validation for vallm
generate_for_vallm(Path("."), depth=3)

# Pre-deployment validation
passed, issues = pre_deploy_check(Path("."))
if not passed:
    print("Deployment blocked:", issues)

Tests

pytest tests/ -v

License