- Reorder sprints for visual-first development - Dashboard (Sprint 4) now follows Physics Engine (Sprint 3) - Infrastructure layers (SCPI, TCP, HAL) follow visual demo - Update project references to py-dvt-ate
65 KiB
Technical Design Document
py_dvt_ate: Implementation Specification
| Document ID | TDD-001 |
|---|---|
| Version | 1.1.0 |
| Status | Draft |
| Author | Kai Chappell |
| Created | 2025-12-01 |
| Last Updated | 2025-12-01 |
Purpose
This document specifies how to implement the py_dvt_ate system. It contains technical details including architecture diagrams, code structures, interfaces, schemas, and specifications.
For what the system must do, see 01_requirements.md.
For why decisions were made, see 03_architecture_decisions.md.
Related Documents
| Document | Purpose |
|---|---|
01_requirements.md |
Defines what the system must do |
02_technical_specification.md |
Specifies how to implement (this document) |
03_architecture_decisions.md |
Explains why decisions were made |
Table of Contents
- System Architecture
- Project Structure
- Module Specifications
- Interface Definitions
- SCPI Protocol Specification
- Physics Model Specification
- Data Schemas
- Configuration Schema
- API Specification
- Development Phases
1. System Architecture
1.1 System Context Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ SYSTEM BOUNDARY │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Simulation Server │ │
│ │ (Separate Process) │ │
│ │ │ │
│ │ Physics Engine ◄───► Virtual Instruments │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ TCP/IP + SCPI │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Test Application │ │
│ │ (Main Process) │ │
│ │ │ │
│ │ CLI / API ───► Test Executive ───► HAL ───► Drivers │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ Data Persistence │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────┐
│ External Actors │
│ │
│ • DVT Engineer (CLI/API) │
│ • File System (Data) │
│ • Real Instruments (Future)│
└─────────────────────────────┘
1.2 Layer Architecture
┌─────────────────────────────────────────────────────────────────────────┐
│ PRESENTATION LAYER │
│ │
│ Components: CLI (Typer), REST API (FastAPI), Dashboard (Streamlit) │
│ Depends on: Application Layer │
├─────────────────────────────────────────────────────────────────────────┤
│ APPLICATION LAYER │
│ │
│ Components: Test Executive, Sequencer, Reporter │
│ Depends on: Domain Layer │
├─────────────────────────────────────────────────────────────────────────┤
│ DOMAIN LAYER │
│ │
│ Components: Test Definitions, Measurement Models, Limit Checking │
│ Depends on: HAL Interfaces (abstractions only) │
├─────────────────────────────────────────────────────────────────────────┤
│ HARDWARE ABSTRACTION LAYER │
│ │
│ Components: IThermalChamber, IPowerSupply, IMultimeter (Protocols) │
│ Depends on: None (pure interfaces) │
├─────────────────────────────────────────────────────────────────────────┤
│ INFRASTRUCTURE LAYER │
│ │
│ Components: Drivers, Transport, Repository, File I/O │
│ Depends on: HAL Interfaces (implements them) │
└─────────────────────────────────────────────────────────────────────────┘
1.3 Component Diagram
┌─────────────────────────────────────────────────────────────────────────────┐
│ TEST APPLICATION │
│ │
│ ┌──────────────┐ ┌──────────────────────────────────────────────────┐ │
│ │ CLI │────▶│ Test Executive │ │
│ │ (Typer) │ │ ┌────────────┐ ┌──────────┐ ┌───────────────┐ │ │
│ └──────────────┘ │ │ Sequencer │ │ Logger │ │ Limit Checker │ │ │
│ │ └────────────┘ └──────────┘ └───────────────┘ │ │
│ ┌──────────────┐ └──────────────────────┬───────────────────────────┘ │
│ │ Streamlit │ │ │
│ │ Dashboard │────────────────────────────┤ │
│ └──────────────┘ ▼ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Hardware Abstraction Layer │ │
│ │ │ │
│ │ ┌────────────────┐ ┌────────────────┐ │ │
│ │ │ IThermalChamber│ │ IPowerSupply │ ... │ │
│ │ └───────┬────────┘ └───────┬────────┘ │ │
│ └──────────┼──────────────────┼────────────────────┘ │
│ │ │ │
│ ┌──────────▼──────────────────▼────────────────────┐ │
│ │ Driver Layer │ │
│ │ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ChamberDriver │ │ PSU Driver │ ... │ │
│ │ └──────┬───────┘ └──────┬───────┘ │ │
│ └─────────┼────────────────┼───────────────────────┘ │
│ │ │ │
│ ┌─────────▼────────────────▼───────────────────────┐ │
│ │ Transport Layer │ │
│ │ (SCPI over TCP/IP) │ │
│ └─────────────────────┬────────────────────────────┘ │
│ │ │
└─────────────────────────────────────────────┼────────────────────────────────┘
│ TCP/IP
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ SIMULATION SERVER │
│ │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ Instrument Servers │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Thermal │ │ Power │ │ DMM │ │ │
│ │ │ Chamber │ │ Supply │ │ Server │ │ │
│ │ │ :5001 │ │ :5002 │ │ :5003 │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ │ │ │ │ │
│ │ └─────────────────┼─────────────────┘ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────┐ │ │
│ │ │ Coupled Physics Engine │ │ │
│ │ │ ┌─────────────────────────┐ │ │ │
│ │ │ │ DUT Thermal Model │ │ │ │
│ │ │ │ DUT Electrical Model │ │ │ │
│ │ │ │ Environment Model │ │ │ │
│ │ │ └─────────────────────────┘ │ │ │
│ │ └─────────────────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
1.4 Deployment Diagram
┌─────────────────────────────────────────────────────────────────┐
│ Docker Compose │
│ │
│ ┌────────────────────────┐ ┌────────────────────────────┐ │
│ │ simulation-server │ │ test-application │ │
│ │ │ │ │ │
│ │ - Physics Engine │ │ - CLI │ │
│ │ - Instrument Servers │ │ - REST API │ │
│ │ - DUT Models │ │ - Test Executive │ │
│ │ │ │ - Streamlit Dashboard │ │
│ │ Ports: 5001-5003 │ │ Ports: 8000, 8501 │ │
│ └────────────────────────┘ └────────────────────────────┘ │
│ │ │ │
│ └───────────────────────────┘ │
│ Internal Network │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Volumes │ │
│ │ ./data/results.db ./data/measurements/ │ │
│ │ ./config/ ./reports/ │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
2. Project Structure
2.1 Directory Layout
thermaulate/
├── pyproject.toml # Project metadata and dependencies
├── README.md # Project overview and quick start
├── CHANGELOG.md # Version history
│
├── docs/
│ ├── 01_requirements.md # Business Requirements
│ ├── 02_technical_specification.md # Technical Design (this doc)
│ └── 03_architecture_decisions.md # Architecture Decisions
│
├── src/
│ └── thermaulate/
│ ├── __init__.py
│ ├── py.typed # PEP 561 marker
│ │
│ ├── physics/ # Physics simulation engine
│ │ ├── __init__.py
│ │ ├── engine.py # Main physics loop
│ │ ├── thermal.py # Thermal domain model
│ │ ├── electrical.py # Electrical domain model
│ │ └── dut/
│ │ ├── __init__.py
│ │ ├── base.py # DUT base class
│ │ └── ldo.py # LDO voltage regulator model
│ │
│ ├── instruments/ # Virtual instrument implementations
│ │ ├── __init__.py
│ │ ├── base.py # Instrument base class
│ │ ├── scpi_parser.py # SCPI command parser
│ │ ├── thermal_chamber.py # Thermal chamber simulator
│ │ ├── power_supply.py # Power supply simulator
│ │ └── multimeter.py # DMM simulator
│ │
│ ├── server/ # Simulation server
│ │ ├── __init__.py
│ │ ├── tcp_server.py # Async TCP server
│ │ └── main.py # Server entry point
│ │
│ ├── transport/ # Communication layer
│ │ ├── __init__.py
│ │ ├── base.py # Transport protocol
│ │ ├── tcp.py # TCP/IP implementation
│ │ └── async_tcp.py # Async TCP implementation
│ │
│ ├── drivers/ # Instrument SCPI drivers
│ │ ├── __init__.py
│ │ ├── base.py # Driver base class
│ │ ├── thermal_chamber.py # Chamber SCPI driver
│ │ ├── power_supply.py # PSU SCPI driver
│ │ └── multimeter.py # DMM SCPI driver
│ │
│ ├── hal/ # Hardware Abstraction Layer
│ │ ├── __init__.py
│ │ ├── interfaces.py # Protocol definitions
│ │ ├── factory.py # Instrument factory
│ │ └── impl/ # HAL implementations
│ │ ├── __init__.py
│ │ ├── thermal_chamber.py
│ │ ├── power_supply.py
│ │ └── multimeter.py
│ │
│ ├── executive/ # Test execution framework
│ │ ├── __init__.py
│ │ ├── sequencer.py # Test sequencer
│ │ ├── context.py # Test context
│ │ ├── logger.py # Test logger
│ │ ├── limits.py # Limit checker
│ │ └── models.py # Domain models
│ │
│ ├── tests/ # DVT test implementations
│ │ ├── __init__.py
│ │ ├── base.py # Test base class
│ │ ├── tempco.py # TempCo characterisation
│ │ └── load_regulation.py # Load regulation test
│ │
│ ├── data/ # Data persistence
│ │ ├── __init__.py
│ │ ├── repository.py # Data access layer
│ │ ├── models.py # Data models
│ │ └── migrations/ # Schema migrations
│ │
│ ├── reporting/ # Report generation (Phase 3)
│ │ ├── __init__.py
│ │ ├── generator.py # Report generator
│ │ ├── pdf.py # PDF output
│ │ ├── html.py # HTML output
│ │ └── templates/ # Report templates
│ │
│ ├── api/ # REST API (Phase 2)
│ │ ├── __init__.py
│ │ ├── main.py # FastAPI app
│ │ └── routes/
│ │ ├── __init__.py
│ │ ├── instruments.py
│ │ ├── tests.py
│ │ └── runs.py
│ │
│ ├── dashboard/ # Streamlit dashboard
│ │ ├── __init__.py
│ │ ├── app.py # Main Streamlit app
│ │ ├── pages/ # Multi-page app
│ │ │ ├── 01_instruments.py
│ │ │ ├── 02_run_test.py
│ │ │ └── 03_results.py
│ │ └── components/ # Reusable UI components
│ │ ├── __init__.py
│ │ └── instrument_panel.py
│ │
│ ├── cli/ # Command-line interface
│ │ ├── __init__.py
│ │ └── main.py # Typer CLI app
│ │
│ └── config/ # Configuration handling
│ ├── __init__.py
│ ├── models.py # Pydantic config models
│ └── loader.py # Config file loader
│
├── tests/ # Test suite
│ ├── conftest.py # pytest fixtures
│ ├── unit/
│ │ ├── test_physics_engine.py
│ │ ├── test_scpi_parser.py
│ │ ├── test_thermal_model.py
│ │ └── ...
│ └── integration/
│ ├── test_instrument_communication.py
│ ├── test_tempco_sequence.py
│ └── ...
│
├── config/ # Configuration files
│ ├── default.yaml # Default configuration
│ └── example_pyvisa.yaml # Example for real hardware
│
├── docker/
│ ├── Dockerfile.server # Simulation server image
│ ├── Dockerfile.app # Test application image
│ └── docker-compose.yml # Full stack orchestration
│
└── scripts/
├── demo.py # Demo script
└── run_tempco.py # Example test execution
2.2 Package Dependencies
thermaulate/
├── cli/ ──────────────────────────────────────────────┐
├── api/ ──────────────────────────────────────────────┤
├── dashboard/ ──────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ PRESENTATION │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
├── executive/ ◄───────────────────────────────────────────────┤
├── tests/ ◄───────────────────────────────────────────────┤
├── reporting/ ◄───────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ APPLICATION │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
├── hal/interfaces ◄───────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ DOMAIN (Abstractions) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ implements│ │
│ ▼ │
├── hal/impl ◄───────────────────────────────────────────────┤
├── drivers/ ◄───────────────────────────────────────────────┤
├── transport/ ◄───────────────────────────────────────────────┤
├── data/ ◄───────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ INFRASTRUCTURE │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
SIMULATION SERVER (Separate Process):
├── physics/ ◄─── Pure domain logic, no external dependencies
├── instruments/ ◄─── Depends on physics
└── server/ ◄─── Depends on instruments
3. Module Specifications
3.1 Physics Module
Responsibility: Simulate coupled thermal-electrical behaviour.
Key Components:
| Component | File | Purpose |
|---|---|---|
| PhysicsEngine | engine.py |
Main simulation loop, state management |
| ThermalModel | thermal.py |
Heat transfer calculations |
| ElectricalModel | electrical.py |
Current/voltage relationships |
| DUTBase | dut/base.py |
Abstract DUT interface |
| LDOModel | dut/ldo.py |
LDO voltage regulator implementation |
State Management:
- Engine maintains global simulation time
- State updates at fixed timestep (default 10ms = 100Hz)
- Immutable state snapshots returned to callers
3.2 Instruments Module
Responsibility: SCPI-compliant virtual instrument behaviour.
Key Components:
| Component | File | Purpose |
|---|---|---|
| InstrumentBase | base.py |
Common instrument functionality |
| SCPIParser | scpi_parser.py |
Parse SCPI command strings |
| ThermalChamberSim | thermal_chamber.py |
Chamber simulation |
| PowerSupplySim | power_supply.py |
PSU simulation |
| MultimeterSim | multimeter.py |
DMM simulation |
Command Processing Flow:
SCPI String → Parser → Command Object → Instrument Handler → Response
3.3 Transport Module
Responsibility: Low-level communication.
Key Components:
| Component | File | Purpose |
|---|---|---|
| Transport Protocol | base.py |
Abstract transport interface |
| TCPTransport | tcp.py |
Synchronous TCP implementation |
| AsyncTCPTransport | async_tcp.py |
Async TCP implementation |
3.4 Drivers Module
Responsibility: Instrument-specific SCPI command sets.
Key Components:
| Component | File | Purpose |
|---|---|---|
| DriverBase | base.py |
Common driver functionality |
| ThermalChamberDriver | thermal_chamber.py |
Chamber SCPI commands |
| PowerSupplyDriver | power_supply.py |
PSU SCPI commands |
| MultimeterDriver | multimeter.py |
DMM SCPI commands |
3.5 HAL Module
Responsibility: Hardware abstraction interfaces.
Key Components:
| Component | File | Purpose |
|---|---|---|
| Protocols | interfaces.py |
Abstract interfaces |
| InstrumentFactory | factory.py |
Creates instrument sets from config |
| HAL Implementations | impl/*.py |
Concrete HAL classes |
3.6 Executive Module
Responsibility: Test execution orchestration.
Key Components:
| Component | File | Purpose |
|---|---|---|
| TestSequencer | sequencer.py |
Run test sequences |
| TestContext | context.py |
Runtime context |
| TestLogger | logger.py |
Measurement logging |
| LimitChecker | limits.py |
Pass/fail evaluation |
| Domain Models | models.py |
Measurement, Result, etc. |
3.7 Dashboard Module
Responsibility: Real-time visualisation via Streamlit.
Key Components:
| Component | File | Purpose |
|---|---|---|
| Main App | app.py |
Streamlit application entry point |
| Instruments Page | pages/01_instruments.py |
Live instrument status |
| Run Test Page | pages/02_run_test.py |
Test execution interface |
| Results Page | pages/03_results.py |
Historical results viewer |
| Instrument Panel | components/instrument_panel.py |
Reusable instrument display |
4. Interface Definitions
4.1 HAL Interfaces
# thermaulate/hal/interfaces.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class IThermalChamber(Protocol):
"""Hardware abstraction for thermal chambers."""
def set_temperature(self, setpoint: float) -> None:
"""Set target temperature in degrees Celsius."""
...
def get_temperature(self) -> float:
"""Get current actual temperature in degrees Celsius."""
...
def get_setpoint(self) -> float:
"""Get current temperature setpoint."""
...
def is_stable(self) -> bool:
"""Check if temperature has stabilised at setpoint."""
...
def wait_until_stable(
self,
timeout: float = 300.0,
poll_interval: float = 1.0
) -> bool:
"""
Block until temperature stabilises or timeout.
Returns:
True if stable, False if timeout
"""
...
def set_ramp_rate(self, rate: float) -> None:
"""Set temperature ramp rate in degrees C per minute."""
...
@runtime_checkable
class IPowerSupply(Protocol):
"""Hardware abstraction for programmable power supplies."""
def set_voltage(self, channel: int, voltage: float) -> None:
"""Set output voltage for specified channel."""
...
def get_voltage(self, channel: int) -> float:
"""Get voltage setpoint for specified channel."""
...
def set_current_limit(self, channel: int, current: float) -> None:
"""Set current limit for specified channel."""
...
def get_current_limit(self, channel: int) -> float:
"""Get current limit for specified channel."""
...
def measure_voltage(self, channel: int) -> float:
"""Measure actual output voltage."""
...
def measure_current(self, channel: int) -> float:
"""Measure actual output current."""
...
def enable_output(self, channel: int, enable: bool) -> None:
"""Enable or disable channel output."""
...
def is_output_enabled(self, channel: int) -> bool:
"""Check if channel output is enabled."""
...
@runtime_checkable
class IMultimeter(Protocol):
"""Hardware abstraction for digital multimeters."""
def measure_dc_voltage(self, range: str = "AUTO") -> float:
"""Measure DC voltage. Range: AUTO, 0.1, 1, 10, 100, 1000."""
...
def measure_dc_current(self, range: str = "AUTO") -> float:
"""Measure DC current."""
...
def measure_resistance(self, range: str = "AUTO") -> float:
"""Measure resistance."""
...
def set_integration_time(self, nplc: float) -> None:
"""Set integration time in power line cycles (0.1 to 100)."""
...
@runtime_checkable
class ITestLogger(Protocol):
"""Abstraction for test data logging."""
def log_measurement(
self,
parameter: str,
value: float,
unit: str,
conditions: dict[str, float] | None = None
) -> None:
"""Log a single measurement."""
...
def log_result(
self,
parameter: str,
value: float,
unit: str,
lower_limit: float | None = None,
upper_limit: float | None = None
) -> None:
"""Log a test result with optional limits."""
...
def log_event(self, message: str, level: str = "INFO") -> None:
"""Log a test event or message."""
...
4.2 Transport Interface
# thermaulate/transport/base.py
from typing import Protocol
class Transport(Protocol):
"""Abstract transport interface for instrument communication."""
def connect(self) -> None:
"""Establish connection to instrument."""
...
def disconnect(self) -> None:
"""Close connection to instrument."""
...
def write(self, command: str) -> None:
"""Send command to instrument."""
...
def read(self, timeout: float | None = None) -> str:
"""Read response from instrument."""
...
def query(self, command: str, timeout: float | None = None) -> str:
"""Send command and read response."""
...
@property
def is_connected(self) -> bool:
"""Check if connection is active."""
...
4.3 Test Interface
# thermaulate/executive/models.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Protocol
from uuid import UUID
class TestStatus(Enum):
PENDING = "pending"
RUNNING = "running"
PASSED = "passed"
FAILED = "failed"
ERROR = "error"
SKIPPED = "skipped"
@dataclass(frozen=True)
class Measurement:
"""Immutable measurement record."""
timestamp: datetime
parameter: str
value: float
unit: str
conditions: dict[str, float] = field(default_factory=dict)
@dataclass(frozen=True)
class TestResult:
"""Immutable test result with limits."""
parameter: str
value: float
unit: str
lower_limit: float | None = None
upper_limit: float | None = None
@property
def passed(self) -> bool | None:
"""Evaluate pass/fail. None if no limits defined."""
if self.lower_limit is None and self.upper_limit is None:
return None
lower_ok = self.lower_limit is None or self.value >= self.lower_limit
upper_ok = self.upper_limit is None or self.value <= self.upper_limit
return lower_ok and upper_ok
@dataclass
class TestContext:
"""Runtime context for test execution."""
run_id: UUID
instruments: "InstrumentSet"
logger: "ITestLogger"
config: dict
class ITest(Protocol):
"""Interface for test implementations."""
@property
def name(self) -> str:
"""Test name identifier."""
...
@property
def description(self) -> str:
"""Human-readable test description."""
...
def execute(self, context: TestContext) -> TestStatus:
"""Execute the test, return status."""
...
4.4 Factory Interface
# thermaulate/hal/factory.py
from dataclasses import dataclass
from typing import Literal
from thermaulate.hal.interfaces import IThermalChamber, IPowerSupply, IMultimeter
@dataclass
class InstrumentSet:
"""Container for all instruments."""
chamber: IThermalChamber
psu: IPowerSupply
dmm: IMultimeter
@dataclass
class InstrumentConfig:
"""Configuration for instrument connections."""
backend: Literal["simulator", "pyvisa"]
# Simulator settings
simulator_host: str = "localhost"
chamber_port: int = 5001
psu_port: int = 5002
dmm_port: int = 5003
# PyVISA settings (for real hardware)
chamber_visa: str | None = None
psu_visa: str | None = None
dmm_visa: str | None = None
class InstrumentFactory:
"""Factory for creating instrument sets from configuration."""
@staticmethod
def create(config: InstrumentConfig) -> InstrumentSet:
"""Create instrument set based on configuration."""
if config.backend == "simulator":
return InstrumentFactory._create_simulated(config)
elif config.backend == "pyvisa":
return InstrumentFactory._create_pyvisa(config)
else:
raise ValueError(f"Unknown backend: {config.backend}")
@staticmethod
def _create_simulated(config: InstrumentConfig) -> InstrumentSet:
"""Create simulated instruments."""
from thermaulate.transport.tcp import TCPTransport
from thermaulate.drivers.thermal_chamber import ThermalChamberDriver
from thermaulate.drivers.power_supply import PowerSupplyDriver
from thermaulate.drivers.multimeter import MultimeterDriver
from thermaulate.hal.impl.thermal_chamber import ThermalChamberHAL
from thermaulate.hal.impl.power_supply import PowerSupplyHAL
from thermaulate.hal.impl.multimeter import MultimeterHAL
chamber_transport = TCPTransport(config.simulator_host, config.chamber_port)
psu_transport = TCPTransport(config.simulator_host, config.psu_port)
dmm_transport = TCPTransport(config.simulator_host, config.dmm_port)
return InstrumentSet(
chamber=ThermalChamberHAL(ThermalChamberDriver(chamber_transport)),
psu=PowerSupplyHAL(PowerSupplyDriver(psu_transport)),
dmm=MultimeterHAL(MultimeterDriver(dmm_transport)),
)
@staticmethod
def _create_pyvisa(config: InstrumentConfig) -> InstrumentSet:
"""Create PyVISA instruments for real hardware."""
# Implementation would use pyvisa.ResourceManager
raise NotImplementedError("PyVISA backend not yet implemented")
5. SCPI Protocol Specification
5.1 Common Commands (IEEE 488.2)
All instruments implement these standard commands:
| Command | Response | Description |
|---|---|---|
*IDN? |
<manufacturer>,<model>,<serial>,<version> |
Identity query |
*RST |
- | Reset to default state |
*CLS |
- | Clear status registers |
*OPC? |
1 |
Operation complete query |
*OPC |
- | Set OPC bit when complete |
SYST:ERR? |
<code>,"<message>" |
Query error queue |
5.2 Thermal Chamber Commands
| Command | Parameters | Response | Description |
|---|---|---|---|
TEMP:SETPOINT |
<float> |
- | Set target temperature (°C) |
TEMP:SETPOINT? |
- | <float> |
Query temperature setpoint |
TEMP:ACTUAL? |
- | <float> |
Query actual temperature |
TEMP:RAMP:RATE |
<float> |
- | Set ramp rate (°C/min) |
TEMP:RAMP:RATE? |
- | <float> |
Query ramp rate |
TEMP:STAB:WIN |
<float> |
- | Set stability window (±°C) |
TEMP:STAB:TIME |
<float> |
- | Set stability time (seconds) |
TEMP:STAB? |
- | 0 or 1 |
Query stability status |
Example Session:
> *IDN?
< py_dvt_ate,VirtualChamber,SN001,1.0.0
> TEMP:SETPOINT 85.0
> TEMP:SETPOINT?
< 85.0
> TEMP:ACTUAL?
< 27.3
> TEMP:STAB?
< 0
5.3 Power Supply Commands
| Command | Parameters | Response | Description |
|---|---|---|---|
INST:SEL |
CH1 or CH2 |
- | Select active channel |
INST:SEL? |
- | CH1 or CH2 |
Query selected channel |
VOLT |
<float> |
- | Set voltage (selected channel) |
VOLT? |
- | <float> |
Query voltage setpoint |
CURR |
<float> |
- | Set current limit |
CURR? |
- | <float> |
Query current limit |
MEAS:VOLT? |
- | <float> |
Measure actual voltage |
MEAS:CURR? |
- | <float> |
Measure actual current |
MEAS:POW? |
- | <float> |
Measure power (V × I) |
OUTP |
ON or OFF or 0 or 1 |
- | Enable/disable output |
OUTP? |
- | 0 or 1 |
Query output state |
Example Session:
> *IDN?
< py_dvt_ate,VirtualPSU,SN002,1.0.0
> INST:SEL CH1
> VOLT 5.0
> CURR 0.5
> OUTP ON
> MEAS:VOLT?
< 4.998
> MEAS:CURR?
< 0.052
5.4 Multimeter Commands
| Command | Parameters | Response | Description |
|---|---|---|---|
CONF:VOLT:DC |
[<range>] |
- | Configure DC voltage measurement |
CONF:CURR:DC |
[<range>] |
- | Configure DC current measurement |
CONF:RES |
[<range>] |
- | Configure resistance measurement |
MEAS:VOLT:DC? |
[<range>] |
<float> |
Measure DC voltage |
MEAS:CURR:DC? |
[<range>] |
<float> |
Measure DC current |
MEAS:RES? |
[<range>] |
<float> |
Measure resistance |
READ? |
- | <float> |
Read with current config |
SENS:VOLT:DC:NPLC |
<float> |
- | Set integration time (PLCs) |
SENS:VOLT:DC:NPLC? |
- | <float> |
Query integration time |
Range Values: AUTO, 0.1, 1, 10, 100, 1000
Example Session:
> *IDN?
< py_dvt_ate,VirtualDMM,SN003,1.0.0
> CONF:VOLT:DC AUTO
> SENS:VOLT:DC:NPLC 10
> MEAS:VOLT:DC?
< 3.2987
5.5 SCPI Parser Specification
# thermaulate/instruments/scpi_parser.py
from dataclasses import dataclass
@dataclass
class SCPICommand:
"""Parsed SCPI command."""
header: str # e.g., "TEMP:SETPOINT" or "*IDN"
arguments: list[str] # e.g., ["85.0"] or []
is_query: bool # True if ends with '?'
@property
def keyword(self) -> str:
"""Return the command keyword without '?'."""
return self.header.rstrip('?')
class SCPIParser:
"""Parse SCPI command strings."""
def parse(self, command_string: str) -> SCPICommand:
"""
Parse a SCPI command string.
Examples:
"*IDN?" -> SCPICommand("*IDN", [], True)
"VOLT 3.3" -> SCPICommand("VOLT", ["3.3"], False)
"TEMP:SETPOINT?" -> SCPICommand("TEMP:SETPOINT", [], True)
"""
command_string = command_string.strip()
is_query = command_string.endswith('?')
# Split into header and arguments
parts = command_string.split(None, 1) # Split on first whitespace
header = parts[0]
arguments = []
if len(parts) > 1:
# Parse comma-separated arguments
arg_string = parts[1]
arguments = [arg.strip() for arg in arg_string.split(',')]
return SCPICommand(
header=header,
arguments=arguments,
is_query=is_query
)
6. Physics Model Specification
6.1 Thermal Model
State Variables:
T_chamber: Chamber air temperature (°C)T_case: DUT case temperature (°C)T_junction: DUT junction temperature (°C)
Parameters:
| Parameter | Symbol | Typical Value | Unit | Description |
|---|---|---|---|---|
| Chamber time constant | τ_chamber | 30 | s | Thermal response time |
| Case time constant | τ_case | 5 | s | Package thermal response |
| Junction time constant | τ_junction | 0.5 | s | Die thermal response |
| Thermal resistance (junction-case) | θ_jc | 15 | °C/W | Junction to case |
| Thermal resistance (case-ambient) | θ_ca | 5 | °C/W | Case to ambient |
Differential Equations:
Chamber temperature (first-order response to setpoint):
dT_chamber/dt = (T_setpoint - T_chamber) / τ_chamber
Case temperature (driven by chamber and self-heating):
dT_case/dt = (T_chamber - T_case + P_diss × θ_ca) / τ_case
Junction temperature (instantaneous, no thermal mass at die level):
T_junction = T_case + P_diss × θ_jc
6.2 LDO Electrical Model
State Variables:
V_in: Input voltage (V)I_load: Load current (A)V_out: Output voltage (V)I_q: Quiescent current (A)
Temperature-Dependent Parameters:
| Parameter | Formula | Description |
|---|---|---|
| Output voltage | V_out(T) = V_nom × (1 + TC_vout × (T - 25) × 1e-6) |
TC in ppm/°C |
| Quiescent current | I_q(T) = I_q_25 × (1 + TC_iq × (T - 25)) |
TC as ratio/°C |
| Dropout voltage | V_do(T) = V_do_25 × (T / 300)^1.5 |
Increases with temp |
Power Dissipation:
P_diss = (V_in - V_out) × I_load + V_in × I_q
LDO Model Parameters:
| Parameter | Symbol | Default Value | Unit |
|---|---|---|---|
| Nominal output voltage | V_nom | 3.3 | V |
| Output voltage TempCo | TC_vout | 50 | ppm/°C |
| Quiescent current at 25°C | I_q_25 | 50 | µA |
| Quiescent current TempCo | TC_iq | 0.003 | 1/°C |
| Dropout voltage at 25°C | V_do_25 | 0.3 | V |
| Max output current | I_max | 0.5 | A |
6.3 Physics Engine Implementation
# thermaulate/physics/engine.py
from dataclasses import dataclass
@dataclass(frozen=True)
class ThermalState:
"""Immutable thermal state snapshot."""
chamber_temperature: float # °C
case_temperature: float # °C
junction_temperature: float # °C
timestamp: float # seconds since start
@dataclass(frozen=True)
class ElectricalState:
"""Immutable electrical state snapshot."""
input_voltage: float # V
output_voltage: float # V
load_current: float # A
quiescent_current: float # A
power_dissipation: float # W
class PhysicsEngine:
"""
Coupled thermal-electrical physics simulation.
Runs at fixed timestep, updating thermal and electrical state.
"""
def __init__(
self,
update_rate_hz: float = 100.0,
dut_model: "DUTModel" = None
):
self.dt = 1.0 / update_rate_hz
self.dut = dut_model or LDOModel()
# Thermal parameters
self.tau_chamber = 30.0 # seconds
self.tau_case = 5.0 # seconds
self.theta_jc = 15.0 # °C/W
self.theta_ca = 5.0 # °C/W
# State
self._t_setpoint = 25.0
self._t_chamber = 25.0
self._t_case = 25.0
self._v_in = 0.0
self._i_load = 0.0
self._output_enabled = False
self._sim_time = 0.0
def step(self) -> None:
"""Advance simulation by one timestep."""
# Update chamber temperature (first-order response)
dT_chamber = (self._t_setpoint - self._t_chamber) / self.tau_chamber
self._t_chamber += dT_chamber * self.dt
# Calculate power dissipation
if self._output_enabled:
p_diss = self._get_power_dissipation()
else:
p_diss = 0.0
# Update case temperature
dT_case = (self._t_chamber - self._t_case + p_diss * self.theta_ca) / self.tau_case
self._t_case += dT_case * self.dt
self._sim_time += self.dt
def get_thermal_state(self) -> ThermalState:
"""Get current thermal state."""
t_junction = self._t_case + self._get_power_dissipation() * self.theta_jc
return ThermalState(
chamber_temperature=self._t_chamber,
case_temperature=self._t_case,
junction_temperature=t_junction,
timestamp=self._sim_time
)
def get_electrical_state(self) -> ElectricalState:
"""Get current electrical state."""
t_junction = self._t_case + self._get_power_dissipation() * self.theta_jc
v_out = self.dut.calculate_output_voltage(t_junction) if self._output_enabled else 0.0
i_q = self.dut.calculate_quiescent_current(t_junction) if self._output_enabled else 0.0
p_diss = self._get_power_dissipation()
return ElectricalState(
input_voltage=self._v_in,
output_voltage=v_out,
load_current=self._i_load if self._output_enabled else 0.0,
quiescent_current=i_q,
power_dissipation=p_diss
)
def set_chamber_setpoint(self, temperature: float) -> None:
"""Set chamber target temperature."""
self._t_setpoint = temperature
def set_input_voltage(self, voltage: float) -> None:
"""Set DUT input voltage."""
self._v_in = voltage
def set_load_current(self, current: float) -> None:
"""Set DUT load current."""
self._i_load = current
def set_output_enabled(self, enabled: bool) -> None:
"""Enable or disable DUT power."""
self._output_enabled = enabled
def _get_power_dissipation(self) -> float:
"""Calculate current power dissipation."""
if not self._output_enabled:
return 0.0
t_junction = self._t_case # Approximate for calculation
return self.dut.calculate_power_dissipation(
self._v_in, self._i_load, t_junction
)
7. Data Schemas
7.1 SQLite Schema (Metadata)
-- File: data/migrations/001_initial.sql
-- Test run metadata
CREATE TABLE IF NOT EXISTS test_runs (
id TEXT PRIMARY KEY, -- UUID
test_name TEXT NOT NULL,
description TEXT,
started_at TEXT NOT NULL, -- ISO8601 timestamp
completed_at TEXT, -- ISO8601 timestamp
status TEXT NOT NULL DEFAULT 'pending', -- pending, running, passed, failed, error
config_json TEXT NOT NULL, -- Test configuration as JSON
operator TEXT,
notes TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
-- Scalar test results with limits
CREATE TABLE IF NOT EXISTS test_results (
id TEXT PRIMARY KEY, -- UUID
test_run_id TEXT NOT NULL,
parameter TEXT NOT NULL,
value REAL NOT NULL,
unit TEXT,
lower_limit REAL,
upper_limit REAL,
passed INTEGER NOT NULL, -- 0 or 1
measured_at TEXT NOT NULL, -- ISO8601 timestamp
FOREIGN KEY (test_run_id) REFERENCES test_runs(id)
);
-- Indexes
CREATE INDEX IF NOT EXISTS idx_test_runs_status ON test_runs(status);
CREATE INDEX IF NOT EXISTS idx_test_runs_name ON test_runs(test_name);
CREATE INDEX IF NOT EXISTS idx_test_results_run ON test_results(test_run_id);
CREATE INDEX IF NOT EXISTS idx_test_results_param ON test_results(parameter);
7.2 Parquet Schema (Time-Series Measurements)
File: data/measurements/run_<uuid>/measurements.parquet
Schema:
├── timestamp: float64 # Seconds since epoch (high precision)
├── parameter: string # Measurement parameter name
├── value: float64 # Measured value
├── unit: string # Unit of measurement
├── temperature: float64 # Chamber temperature at measurement
├── input_voltage: float64 # DUT input voltage at measurement
├── load_current: float64 # DUT load current at measurement
7.3 Data Repository Interface
# thermaulate/data/repository.py
from typing import Protocol
from uuid import UUID
class ITestRepository(Protocol):
"""Repository interface for test data."""
def create_run(
self,
test_name: str,
config: dict,
operator: str | None = None
) -> UUID:
"""Create a new test run, return its ID."""
...
def update_run_status(self, run_id: UUID, status: str) -> None:
"""Update test run status."""
...
def complete_run(self, run_id: UUID, status: str) -> None:
"""Mark test run as complete with final status."""
...
def save_result(
self,
run_id: UUID,
parameter: str,
value: float,
unit: str,
lower_limit: float | None = None,
upper_limit: float | None = None
) -> None:
"""Save a test result."""
...
def save_measurements(
self,
run_id: UUID,
measurements: list["Measurement"]
) -> None:
"""Save batch of measurements to Parquet."""
...
def get_run(self, run_id: UUID) -> "TestRun":
"""Get test run by ID."""
...
def get_results(self, run_id: UUID) -> list["TestResult"]:
"""Get all results for a test run."""
...
def get_measurements_dataframe(self, run_id: UUID):
"""Get measurements as pandas DataFrame."""
...
8. Configuration Schema
8.1 Configuration File Structure
# config/default.yaml
# Instrument backend configuration
instruments:
backend: simulator # "simulator" or "pyvisa"
simulator:
host: localhost
thermal_chamber_port: 5001
power_supply_port: 5002
multimeter_port: 5003
pyvisa:
thermal_chamber: "TCPIP::192.168.1.10::5001::SOCKET"
power_supply: "TCPIP::192.168.1.11::5002::SOCKET"
multimeter: "TCPIP::192.168.1.12::5003::SOCKET"
# Physics simulation parameters
physics:
update_rate_hz: 100
thermal:
chamber_time_constant_s: 30.0
case_time_constant_s: 5.0
theta_jc: 15.0 # °C/W
theta_ca: 5.0 # °C/W
chamber:
ramp_rate_c_per_min: 10.0
stability_window_c: 0.5
stability_time_s: 30.0
# DUT model configuration
dut:
model: ldo
parameters:
nominal_output_voltage: 3.3
tempco_ppm_per_c: 50
quiescent_current_ua: 50
quiescent_current_tempco: 0.003
dropout_voltage: 0.3
# Data storage paths
data:
database_path: "./data/thermaulate.db"
measurements_dir: "./data/measurements"
reports_dir: "./data/reports"
# Logging configuration
logging:
level: INFO
file: "./data/logs/thermaulate.log"
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# Dashboard (Streamlit)
dashboard:
enabled: true
port: 8501
# API server (optional, Phase 2)
api:
enabled: false
host: "0.0.0.0"
port: 8000
8.2 Pydantic Configuration Models
# thermaulate/config/models.py
from pydantic import BaseModel, Field
from typing import Literal
class SimulatorConfig(BaseModel):
host: str = "localhost"
thermal_chamber_port: int = 5001
power_supply_port: int = 5002
multimeter_port: int = 5003
class PyVISAConfig(BaseModel):
thermal_chamber: str | None = None
power_supply: str | None = None
multimeter: str | None = None
class InstrumentsConfig(BaseModel):
backend: Literal["simulator", "pyvisa"] = "simulator"
simulator: SimulatorConfig = Field(default_factory=SimulatorConfig)
pyvisa: PyVISAConfig = Field(default_factory=PyVISAConfig)
class ThermalConfig(BaseModel):
chamber_time_constant_s: float = 30.0
case_time_constant_s: float = 5.0
theta_jc: float = 15.0
theta_ca: float = 5.0
class ChamberConfig(BaseModel):
ramp_rate_c_per_min: float = 10.0
stability_window_c: float = 0.5
stability_time_s: float = 30.0
class PhysicsConfig(BaseModel):
update_rate_hz: float = 100.0
thermal: ThermalConfig = Field(default_factory=ThermalConfig)
chamber: ChamberConfig = Field(default_factory=ChamberConfig)
class DUTParameters(BaseModel):
nominal_output_voltage: float = 3.3
tempco_ppm_per_c: float = 50.0
quiescent_current_ua: float = 50.0
quiescent_current_tempco: float = 0.003
dropout_voltage: float = 0.3
class DUTConfig(BaseModel):
model: str = "ldo"
parameters: DUTParameters = Field(default_factory=DUTParameters)
class DataConfig(BaseModel):
database_path: str = "./data/thermaulate.db"
measurements_dir: str = "./data/measurements"
reports_dir: str = "./data/reports"
class LoggingConfig(BaseModel):
level: str = "INFO"
file: str = "./data/logs/thermaulate.log"
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
class DashboardConfig(BaseModel):
enabled: bool = True
port: int = 8501
class APIConfig(BaseModel):
enabled: bool = False
host: str = "0.0.0.0"
port: int = 8000
class AppConfig(BaseModel):
"""Root configuration model."""
instruments: InstrumentsConfig = Field(default_factory=InstrumentsConfig)
physics: PhysicsConfig = Field(default_factory=PhysicsConfig)
dut: DUTConfig = Field(default_factory=DUTConfig)
data: DataConfig = Field(default_factory=DataConfig)
logging: LoggingConfig = Field(default_factory=LoggingConfig)
dashboard: DashboardConfig = Field(default_factory=DashboardConfig)
api: APIConfig = Field(default_factory=APIConfig)
9. API Specification
9.1 REST Endpoints (Phase 2)
| Method | Endpoint | Description |
|---|---|---|
| GET | /api/v1/health |
Health check |
| GET | /api/v1/instruments |
List instrument status |
| GET | /api/v1/instruments/{id} |
Get instrument details |
| GET | /api/v1/tests |
List available tests |
| GET | /api/v1/tests/{name} |
Get test configuration schema |
| POST | /api/v1/runs |
Start a test run |
| GET | /api/v1/runs |
List test runs |
| GET | /api/v1/runs/{id} |
Get test run status |
| GET | /api/v1/runs/{id}/results |
Get test results |
| GET | /api/v1/runs/{id}/measurements |
Get measurements (paginated) |
| DELETE | /api/v1/runs/{id} |
Delete test run |
9.2 WebSocket Events (Phase 2)
| Event | Direction | Payload | Description |
|---|---|---|---|
instrument.status |
Server→Client | {id, connected, state} |
Instrument status update |
run.progress |
Server→Client | {run_id, step, total, message} |
Test progress |
run.measurement |
Server→Client | {run_id, parameter, value} |
New measurement |
run.complete |
Server→Client | {run_id, status, summary} |
Test completed |
10. Development Phases
10.1 Phase 1: Vertical Slice (MVP)
Goal: End-to-end "Virtual Lab Bench" - Physics Engine → HAL → Driver → Streamlit UI.
Deliverables:
- Project scaffolding (pyproject.toml, directory structure)
- Physics engine with LDO DUT model
- Thermal chamber simulator with SCPI
- Power supply simulator with SCPI
- DMM simulator with SCPI
- TCP server for instruments
- Transport layer (TCP client)
- SCPI drivers for all instruments
- HAL interfaces and implementations
- Instrument factory with simulator backend
- Basic CLI for manual instrument control
- Streamlit dashboard with live instrument visualisation
- SQLite + Parquet data persistence
- TempCo characterisation test
- Unit tests for core modules (≥80% coverage)
Acceptance Criteria:
- Can start simulation server with single command
- Can observe instruments in Streamlit dashboard
- Can execute TempCo test via CLI
- Results show temperature-dependent behaviour
- Self-heating effect visible in results
- Physics coupling demonstrated end-to-end
10.2 Phase 2: Test Framework & API
Goal: Complete test executive with REST API.
Deliverables:
- Test sequencer with configuration
- Limit checking engine
- Load Regulation vs Temperature test
- FastAPI REST endpoints
- WebSocket real-time updates
- OpenAPI documentation
- Test configuration via YAML
- CSV export
Acceptance Criteria:
- Can run multi-test sequences
- API fully documented
- Can start tests via API
- Real-time updates in Streamlit
10.3 Phase 3: Reporting & Polish
Goal: Professional reporting and production readiness.
Deliverables:
- PDF report generation
- HTML report generation
- Docker Compose deployment
- GitHub Actions CI/CD
- Comprehensive README
- Architecture documentation
- Demo scripts with sample output
- Screenshots/GIFs for README
Acceptance Criteria:
- Professional reports with charts
- One-command startup (docker-compose up)
- All documentation complete
- Impressive demo scenario
Appendix A: Technology Versions
| Technology | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Runtime |
| NumPy | ≥1.24 | Numerical |
| SciPy | ≥1.11 | Scientific |
| Pydantic | ≥2.0 | Config |
| Typer | ≥0.9 | CLI |
| Streamlit | ≥1.28 | Dashboard |
| FastAPI | ≥0.100 | API (Phase 2) |
| PyArrow | ≥14.0 | Parquet |
| pytest | ≥7.0 | Testing |
| Ruff | ≥0.1 | Linting |
| mypy | ≥1.0 | Types |
Appendix B: Dependencies (pyproject.toml)
[project]
name = "thermaulate"
version = "0.1.0"
description = "Coupled Physics DVT Simulation Platform"
requires-python = ">=3.11"
dependencies = [
"numpy>=1.24",
"scipy>=1.11",
"pydantic>=2.0",
"pyyaml>=6.0",
"typer>=0.9",
"rich>=13.0",
"pyarrow>=14.0",
"streamlit>=1.28",
"pandas>=2.0",
"plotly>=5.18",
]
[project.optional-dependencies]
api = [
"fastapi>=0.100",
"uvicorn>=0.23",
"websockets>=11.0",
]
reports = [
"jinja2>=3.1",
"weasyprint>=60.0",
]
dev = [
"pytest>=7.0",
"pytest-cov>=4.0",
"pytest-asyncio>=0.21",
"ruff>=0.1",
"mypy>=1.0",
]
[project.scripts]
thermaulate = "thermaulate.cli.main:app"
thermaulate-server = "thermaulate.server.main:main"
thermaulate-dashboard = "thermaulate.dashboard.app:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.mypy]
python_version = "3.11"
strict = true
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
End of Technical Design Document