Skip to main content

OqlOS — Operation Query Language runtime for hardware testing

Project description

OqlOS — Operation Query Language Runtime

AI Cost Tracking

PyPI Version Python License AI Cost Human Time Model

  • 🤖 LLM usage: $0.7500 (5 commits)
  • 👤 Human dev: ~$547 (5.5h @ $100/h, 30min dedup)

Generated on 2026-04-15 using openrouter/qwen/qwen3-coder-next


OqlOS is the core runtime for executing OQL (Operation Query Language) hardware testing scenarios. It provides the execution engine, hardware abstraction layer, and API server for running automated hardware tests.

Installation

# Install from source with development dependencies
pip install -e ".[dev]"

# Basic installation
pip install -e .

Requirements

  • Python 3.10+
  • FastAPI, Uvicorn (for API server)
  • Modbus support (for hardware communication)

Quick Start

Start the API Server

# Start with real hardware
oqlos-server --port 8200

# Run with mock hardware (development/testing)
OQLOS_HARDWARE_MODE=mock oqlos-server --port 8200

Run a Scenario

from oqlos.core.interpreter import CqlInterpreter

source = """
SCENARIO: "Test"
DEVICE_TYPE: "BA"
GOAL: Check
  1. Step:
    → Sensor.read AI01
"""

interp = CqlInterpreter(mode="dry-run")
result = interp.run(source, "test.oql")
print(result.ok)  # True if successful

Package Structure

oqlos/
├── core/               # Parser, executor, state machine, interpreter
│   ├── interpreter.py  # CqlInterpreter main execution engine
│   ├── parser.py       # OQL language parser
│   └── cql_parser.py   # Legacy CQL parser support
├── models/             # Data models
│   ├── scenario.py     # Scenario definition models
│   ├── execution.py    # Execution state models
│   └── peripheral.py   # Hardware peripheral models
├── hardware/           # Hardware abstraction
│   ├── gateway.py      # Hardware gateway interface
│   ├── modbus/         # Modbus communication
│   └── drivers/        # Device drivers
├── api/                # REST API
│   ├── server.py       # FastAPI application
│   └── routes/         # API endpoints
├── executor/           # Scenario execution logic
├── scenarios/          # Sample .oql scenario files
└── shared/             # Utilities (logger, config, version)

Core Components

CqlInterpreter

The main execution engine for OQL scenarios:

from oqlos.core.interpreter import CqlInterpreter

# Modes: "dry-run", "execute", "validate"
interp = CqlInterpreter(
    mode="dry-run",
    firmware_url="http://localhost:8202",
    quiet=False
)

result = interp.run(source_code, filename)
# result.ok: bool — execution success
# result.events: list — execution trace
# result.variables: dict — captured variables

Parser

Two-phase parsing pipeline:

  1. Raw Parser — Converts OQL text to structured blocks
  2. CQL Parser — Processes blocks into executable commands
from oqlos.core.parser import parse_scenario
from oqlos.core.cql_parser import CqlParser

blocks = parse_scenario(source)
parser = CqlParser(blocks)
scenario = parser.parse()

API Endpoints

When running oqlos-server:

Endpoint Method Description
/api/hardware/peripherals GET List connected hardware
/api/scenarios GET List available scenarios
/api/scenarios/{id}/run POST Execute a scenario
/health GET Health check

OQL Scenario Format

OQL scenarios define hardware testing procedures declaratively:

SCENARIO: "PSS 7000 Mask Test"
DEVICE_TYPE: "BA"
DEVICE_MODEL: "PSS 7000"
MANUFACTURER: "Dräger"

@Namespace.ScenarioName
  intervals: [tt#000, tt#001]

GOAL: Visual Inspection
  1. Check mask surface:
    → Valve.open NC
    WAIT 2000
    → Sensor.read AI01
    IF [AI01] [>=] [-15 mbar] ELSE ERROR "Pressure too low"
    SAVE: pressure_reading

CONFIG Blocks (Configuration Goals)

Use CONFIG: for hardware initialization and setup procedures. CONFIG blocks are semantically identical to GOAL blocks but marked with [CONFIG] prefix for clarity in logs and documentation.

Basic CONFIG Example

SCENARIO: "System Startup"
DEVICE_TYPE: "BA"

CONFIG: Safety Initialization
  # Always disable pump on startup
  SET 'pump-main' '0'
  SET 'PUMP' 'off'
  WAIT 500

CONFIG: Valve Reset
  # Close all valves to known state
  SET 'valve-nc' 'closed'
  SET 'valve-sc' 'closed'
  SET 'valve-wc' 'closed'
  WAIT 300

GOAL: Pressure Test
  SET 'valve-nc' 'open'
  WAIT 1000
  → Sensor.read AI01
  SAVE: pressure_test

Configuration File: config-peripherals.oql

Full peripheral initialization scenario:

SCENARIO: 'Konfiguracja Peryferii'
DEVICE_TYPE: 'BA'
DEVICE_MODEL: 'PSS 7000'
MANUFACTURER: 'Dräger'

# ============================================
# PUMP INITIALIZATION
# ============================================
CONFIG: INIT Pompa
  SET 'pump-main' '0'
  SET 'pompa 1' '0'
  SET 'PUMP' 'off'
  WAIT 500

# ============================================
# VALVE INITIALIZATION
# ============================================
CONFIG: INIT Zawory NC
  SET 'valve-nc' 'closed'
  SET 'zawór NC' 'closed'
  WAIT 300

CONFIG: INIT Zawory SC
  SET 'valve-sc' 'closed'
  SET 'zawór SC' 'closed'
  WAIT 300

CONFIG: INIT Zawory ogólne
  SET 'valve-1' '0'
  SET 'valve-2' '0'
  SET 'valve-3' '0'
  SET 'valve-4' '0'
  WAIT 500

# ============================================
# SYSTEM READY STATE
# ============================================
CONFIG: STATE Ready
  SAVE: system_ready
  WAIT 1000

Running Configuration

# Dry-run (validate and simulate)
oqlctl run scenarios/config-peripherals.oql --mode dry-run

# Execute on real hardware
oqlctl run scenarios/config-peripherals.oql --mode execute

# Execute with custom firmware URL
oqlctl run scenarios/config-peripherals.oql \
  --firmware-url http://localhost:8202 \
  --mode execute

CLI Output Example

📋 CQL: 'Konfiguracja Peryferii'
🔧 Device: 'BA' / 'PSS 7000'
🎯 GOAL: [CONFIG] INIT Pompa
  📌 Step 0: [CONFIG] INIT Pompa
    ⚙️ SET [pump-main] = [0]
    ⚙️ SET [pompa 1] = [0]
    ⏳ WAIT 0.5s
    ✅ [passed] [CONFIG] INIT Pompa
🎯 GOAL: [CONFIG] INIT Zawory NC
  ⚙️ SET [valve-nc] = [closed]
  ...
✅ 'Konfiguracja Peryferii': 11/11 passed

Supported Hardware

  • Valves: valve-1 through valve-14, valve-nc, valve-sc, valve-wc (Modbus RTU via /dev/ttyACM1 @ 19200 8N1)
  • Pump: pump-main (DRI0050 PWM motor driver via HTTP :49055)
  • Artificial lung: lung-main (Tic T249 stepper via HTTP :8205)
  • Sensors: AI01 (NC), AI02 (SC), AI03 (WC) (piADC ADS1115 via HTTP :8204)

Hardware Adapters

Adapter Class Protocol Default URL
Motor (pump) _DRI0050MotorAdapter HTTP POST /api/speed http://localhost:49055
Lung (artificial lung) _Tic249LungAdapter HTTP POST /api/lung http://localhost:8205
Valves _ModbusAdapter Modbus RTU (pymodbus) /dev/ttyACM1 serial
Sensors _PiAdcAdapter HTTP GET /api/v1/hardware/sensor/{id} http://localhost:8204

Hardware Identification & Diagnostics

The /api/v1/hardware/identify endpoint returns the adapter registry, live probe status, and a diagnostics block with:

  • USB device inventory
  • Serial port inventory (ttyACM* and ttyUSB*)
  • I2C bus inventory (/dev/i2c-*)
  • Best-effort bridge health snapshot for piadc, motor, lung, and modbus

Environment Variables

Variable Default Description
OQLOS_HARDWARE_MODE mock mock or real
MOTOR_URL http://localhost:49055 DRI0050 motor service
LUNG_MOTOR_URL http://localhost:8205 Tic T249 lung service
PIADC_URL http://localhost:8080 piADC sensor service
MODBUS_SERIAL_PORT /dev/ttyACM1 Modbus RTU serial port
MODBUS_BAUD_RATE 19200 Modbus baud rate

Docker Deployment

# Development
docker-compose -f docker/docker-compose.dev.yml up

# Production
docker-compose -f docker/docker-compose.prod.yml up -d

Testing

# Run all tests (96 passing)
pytest

# Run with coverage
pytest --cov=oqlos

# Run specific test file
pytest tests/test_interpreter.py -v

# Run OQL scenarios (dry-run)
python -m oqlos.core.interpreter scenarios/test-pompy.oql --mode dry-run

Status: 96 tests passing, 3 scenarios (12/12 goals), CC̄≤15, 0 violations

Documentation

License

Licensed under Apache-2.0.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

oqlos-0.1.1.tar.gz (92.6 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

oqlos-0.1.1-py3-none-any.whl (101.4 kB view details)

Uploaded Python 3

File details

Details for the file oqlos-0.1.1.tar.gz.

File metadata

  • Download URL: oqlos-0.1.1.tar.gz
  • Upload date:
  • Size: 92.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for oqlos-0.1.1.tar.gz
Algorithm Hash digest
SHA256 1e49a9e2676552347d1dd18042b53112d84fd618003d7432fbdfa1348c374195
MD5 addb78dc60925d19e17121217d8e837f
BLAKE2b-256 c6a7201e9d16684d80223c55414b04febbf348ad6bfc104aedffe9c192ba9719

See more details on using hashes here.

File details

Details for the file oqlos-0.1.1-py3-none-any.whl.

File metadata

  • Download URL: oqlos-0.1.1-py3-none-any.whl
  • Upload date:
  • Size: 101.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for oqlos-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 20ff72af678ffd6ea8be3c6ddc16165dcc7bd00bf5ef02f9f71d243ebed0fc93
MD5 d830d80289219897ab27acba308e5f20
BLAKE2b-256 2d57aeaadb8696c7bd7bd2f8a62d091cd3d8730cd8458f4b21f7b441a6a54659

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page