# Architecture Source: https://docs.qubit.energy/architecture The six-layer Qubit Foundation stack for energy data infrastructure # Architecture Overview The Qubit Foundation employs a comprehensive six-layer architecture designed to handle the full lifecycle of energy data - from ingestion and processing to optimization and user interaction. Each layer builds upon the previous, creating a robust, scalable infrastructure for the energy transition. ## The Six-Layer Stack **Standardization and Ingestion** The foundational layer that establishes common data models and ingests information from diverse energy sources. * **Schemas**: Universal JSON Schema definitions for all energy data types * **Connectors**: Protocol-specific adapters for MQTT, OCPP, Modbus, and more * **Adapters**: Data transformation, normalization, and validation *Status: In Production* **AI-Powered Forecasting** Production-ready machine learning models that predict energy generation, consumption, and grid conditions. * Solar generation forecasting with weather integration * Demand and load prediction with behavioral patterns * Feature engineering for energy time series * Sub-second inference with confidence intervals *Status: In Production* **Intelligent Resource Management** Mathematical optimization algorithms that determine optimal resource allocation and scheduling. * EV fleet charging schedule optimization with priority and tariff-aware scheduling * Battery storage peak shaving with two-pass dispatch algorithm * Multi-objective optimization balancing cost, carbon, and peak reduction * Native integration with Layer 1 tariff schemas and Layer 2 forecasts *Status: In Production* **Command Dispatch and Asset Orchestration** Converts Layer 3 optimization schedules into real-time device commands with protocol translation and state management. * Dispatch engine converting schedules to timed commands with lifecycle tracking * OCPP-aligned EV charger and battery state machines * Protocol adapters for OCPP (EV chargers) and Modbus (batteries) * Typed event bus for real-time coordination across components *Status: In Production* **Economic Verification and Settlement** Cryptographically secure systems for energy transaction verification and economic settlement. * Blockchain-based energy certificates * Automated payment and settlement * Carbon credit tracking and verification * Renewable energy credit management *Status: Research Phase* **User Experience and APIs** Intuitive interfaces and APIs that make energy data accessible to end users, developers, and applications. * Developer APIs and SDKs * Consumer mobile and web applications * Utility operator dashboards * Third-party integration endpoints *Status: Planning Phase* ## Data Flow ```mermaid theme={null} graph TD A[Energy Assets] --> B[Layer 1: Data] B --> C[Layer 2: Prediction] B --> D[Layer 3: Optimization] C --> D D --> E[Layer 4: Coordination] E --> F[Layer 5: Settlement] F --> G[Layer 6: Interface] G --> H[End Users] subgraph "Real-time Processing" B C D end subgraph "Economic Layer" E F end ``` ## Design Principles Each layer is independently deployable and can be replaced or upgraded without affecting others Designed to handle billions of devices and exabytes of data with horizontal scaling Open standards ensure compatibility across vendors, protocols, and implementations End-to-end encryption, zero-trust architecture, and cryptographic verification Sub-second latency for critical grid operations and millisecond optimization cycles Community-driven development with transparent governance and open licensing ## Technology Stack ### Core Technologies * **Languages**: Python, Rust, TypeScript * **Databases**: InfluxDB, PostgreSQL, Redis * **Protocols**: MQTT, OCPP, Modbus TCP/RTU, DNP3 * **Standards**: JSON Schema, OpenAPI, IEC 61850 * **ML/AI**: TensorFlow, PyTorch, scikit-learn * **Optimization**: CVXPY, OR-Tools, Gurobi * **Streaming**: Apache Kafka, Apache Flink * **Compute**: Kubernetes, Docker, Apache Spark * **Consensus**: Byzantine Fault Tolerance algorithms * **Networking**: libp2p, WebRTC, WebSockets * **Security**: TLS 1.3, Ed25519, AES-256-GCM * **Protocols**: Custom P2P energy trading protocols * **Blockchain**: Ethereum, Polygon, custom L2 solutions * **Smart Contracts**: Solidity, Vyper * **Oracles**: Chainlink, custom price feeds * **Cryptography**: Zero-knowledge proofs, merkle trees ## Deployment Models The Qubit Foundation architecture supports multiple deployment patterns: ### Cloud-Native * Full stack deployment on AWS, GCP, or Azure * Auto-scaling and managed services * Global edge deployment for low latency ### Hybrid * Critical components on-premises * Non-critical processing in the cloud * Data sovereignty compliance ### Edge-First * Local processing for real-time requirements * Minimal cloud dependencies * Offline operation capabilities ### Community * Distributed deployment across participants * Shared infrastructure costs * Collective ownership model ## Integration Patterns The Qubit Foundation is designed for **production energy systems**. Always follow proper safety protocols and testing procedures when integrating with live grid infrastructure. Common integration approaches: 1. **Schema-First**: Start with standardized data models 2. **Connector-Based**: Use protocol-specific adapters 3. **API-Driven**: RESTful and GraphQL interfaces 4. **Event-Streaming**: Real-time data pipelines 5. **Batch Processing**: Historical data analysis ## Performance Characteristics Qubit Foundation Performance Metrics

The architecture is designed to handle: * **Throughput**: 10M+ messages per second per node * **Latency**: Sub-millisecond for critical operations * **Scale**: Billions of connected devices * **Availability**: 99.99% uptime with automatic failover * **Storage**: Exabyte-scale time-series data ## Security Model All Qubit Foundation components implement **zero-trust security** by default, with end-to-end encryption and cryptographic verification at every layer. Security is built into every layer: * **Layer 1**: Device authentication, data encryption, schema validation * **Layer 2**: Model integrity verification, secure computation * **Layer 3**: Optimization result verification, tamper-proof algorithms * **Layer 4**: Byzantine fault tolerance, cryptographic consensus * **Layer 5**: Smart contract audits, economic security mechanisms * **Layer 6**: OAuth 2.0, API rate limiting, user privacy controls ## Next Steps Ready to start building? Begin with Layer 1 Data to establish your foundation: Set up schemas, connectors, and adapters for your energy data Or explore specific layers based on your needs: Universal energy data models Connect any energy asset Transform and validate data # Introduction Source: https://docs.qubit.energy/introduction Building the open energy data infrastructure for a sustainable future # Welcome to Qubit Foundation The Qubit Foundation is pioneering the development of open-source energy data infrastructure that enables the global transition to sustainable energy systems. We're building the foundational technology layer that connects, standardizes, and optimizes energy data across the entire ecosystem. ## Our Mission To accelerate the energy transition by creating universal, open standards and tools that enable seamless integration between energy assets, monitoring systems, and optimization platforms. We believe that open, interoperable data infrastructure is critical to achieving a sustainable energy future. ## The Challenge The energy sector faces unprecedented transformation: Thousands of proprietary protocols and data formats prevent effective integration Distributed energy resources growing from millions to billions of connected devices Grid stability demands millisecond-level coordination across diverse assets Complex market mechanisms require sophisticated settlement and verification ## Our Approach The Qubit Foundation addresses these challenges through a comprehensive six-layer architecture: 1. **Standardized Data Models** - Universal schemas for energy data 2. **Protocol Agnostic Ingestion** - Connect any device or system 3. **Intelligent Processing** - From raw data to actionable insights 4. **Distributed Coordination** - Orchestrate complex energy systems 5. **Transparent Settlement** - Enable new economic models 6. **Intuitive Interfaces** - Make energy data accessible to all ## Why Open Source? The energy transition is too important to be controlled by any single entity. Open source development ensures that solutions are built by the community, for the community, with transparency and collaboration at the core. No vendor lock-in. Organizations can adopt, extend, and customize the infrastructure to meet their specific needs while maintaining interoperability with the broader ecosystem. Open standards and implementations reduce integration costs and time-to-market for new energy solutions, enabling faster deployment of sustainable technologies. Contributors from utilities, technology companies, research institutions, and independent developers work together to solve shared challenges. ## Real-World Impact Our infrastructure powers: * **Grid Operators** managing distributed energy resources at scale * **Energy Communities** coordinating local generation and consumption * **EV Networks** optimizing charging across millions of vehicles * **Industrial Facilities** reducing energy costs through intelligent automation * **Renewable Developers** maximizing asset performance and revenue * **Energy Retailers** offering dynamic, personalized energy services ## Get Started Deploy your first connector in minutes Generate energy predictions with ML models Optimize EV charging and battery dispatch Dispatch schedules to real energy assets ## Core Principles Every specification, schema, and protocol we develop is open and freely available Systems should work together seamlessly, regardless of vendor or technology Built-in security with end-to-end encryption and privacy-preserving architectures Designed to handle billions of devices and exabytes of data Simple, intuitive APIs and comprehensive documentation ## Join Us The energy transition requires unprecedented collaboration. Whether you're a utility, technology provider, researcher, or developer, there's a place for you in the Qubit Foundation community. Contribute code, build integrations, and shape the future of energy data Adopt open standards, reduce integration costs, and accelerate innovation *** *The Qubit Foundation is building the digital infrastructure for the energy transition. Together, we're creating a more sustainable, efficient, and equitable energy future.* # Adapters Source: https://docs.qubit.energy/layer-1/adapters Data transformation, normalization, and validation for energy data # Qubit Energy Adapters Data transformation adapters that normalize, validate, and enrich energy data from diverse sources. Adapters bridge the gap between raw device data and the standardized Qubit Energy Schema format. ## Overview Energy data arrives in countless formats with inconsistent units, timezones, and quality indicators. Qubit Adapters solve this by providing: Automatic conversion between measurement systems (SI, Imperial, custom) Intelligent timezone detection and UTC normalization Multi-layer validation with quality scoring Add metadata, context, and calculated fields ## Core Adapters ### Unit Converter Handles conversion between different measurement systems commonly used in energy applications. ```python theme={null} from qubit.adapters.units import PowerConverter # Convert various power units to standard kW power_kw = PowerConverter.to_kilowatts(5000, "W") # 5.0 kW power_kw = PowerConverter.to_kilowatts(2.5, "MW") # 2500.0 kW power_kw = PowerConverter.to_kilowatts(3.4, "hp") # 2.54 kW # Automatic unit detection normalized = PowerConverter.normalize("1500 Watts") # (1.5, "kW") ``` ```python theme={null} from qubit.adapters.units import EnergyConverter # Convert to standard kWh energy_kwh = EnergyConverter.to_kilowatt_hours(5000, "Wh") # 5.0 kWh energy_kwh = EnergyConverter.to_kilowatt_hours(2.5, "MWh") # 2500.0 kWh energy_kwh = EnergyConverter.to_kilowatt_hours(1.2, "GJ") # 333.33 kWh # Complex conversions with time periods energy_kwh = EnergyConverter.from_rate(150, "kW", "30min") # 75.0 kWh ``` ```python theme={null} from qubit.adapters.units import TemperatureConverter # Convert to Celsius temp_c = TemperatureConverter.to_celsius(77, "F") # 25.0°C temp_c = TemperatureConverter.to_celsius(298.15, "K") # 25.0°C # With precision control temp_c = TemperatureConverter.to_celsius( 98.6, "F", precision=1 ) # 37.0°C ``` ```python theme={null} from qubit.adapters.units import UnitConverter # Register custom unit conversion UnitConverter.register_conversion( "BTU", "kWh", lambda x: x * 0.000293071 ) # Use custom conversion energy_kwh = UnitConverter.convert(10000, "BTU", "kWh") # 2.93 kWh ``` ### Timezone Adapter Handles the complexity of timezone conversion and DST transitions for global energy operations. ```python theme={null} from qubit.adapters.timezone import TimezoneAdapter import datetime # Initialize adapter tz_adapter = TimezoneAdapter() # Convert local time to UTC local_time = datetime.datetime(2024, 7, 15, 14, 30) # 2:30 PM utc_time = tz_adapter.to_utc( local_time, timezone="America/Los_Angeles" ) # 2024-07-15T21:30:00Z # Handle DST transitions automatically winter_time = datetime.datetime(2024, 1, 15, 14, 30) utc_winter = tz_adapter.to_utc( winter_time, timezone="America/Los_Angeles" ) # 2024-01-15T22:30:00Z (PST) # Bulk timezone conversion timestamps = [ "2024-01-15 14:30:00", "2024-07-15 14:30:00" ] utc_timestamps = tz_adapter.batch_convert( timestamps, source_timezone="America/Los_Angeles", source_format="%Y-%m-%d %H:%M:%S" ) ``` ### Data Validation Multi-layer validation ensures data quality and schema compliance: Strict JSON Schema validation against Qubit Energy Schemas ```python theme={null} from qubit.adapters.validation import SchemaValidator validator = SchemaValidator("timeseries", version="0.2") result = validator.validate({ "id": "ts_solar_001", "asset_id": "ast_inverter_001", "metric": "energy_generation", "value": 2500.0, "unit": "kWh", "timestamp": "2024-01-15T14:30:00Z" }) if result.is_valid: print("✅ Data is valid") else: print(f"❌ Validation errors: {result.errors}") ``` Check that values fall within expected ranges for the metric type ```python theme={null} from qubit.adapters.validation import RangeValidator # Define reasonable ranges for metrics ranges = { "energy_generation": {"min": 0, "max": 10000}, # kWh "ambient_temperature": {"min": -40, "max": 60} # °C } validator = RangeValidator(ranges) result = validator.validate("energy_generation", 2500.0) ``` Assign confidence scores based on data characteristics ```python theme={null} from qubit.adapters.validation import QualityScorer scorer = QualityScorer() quality = scorer.score_timeseries({ "value": 2500.0, "timestamp": "2024-01-15T14:30:00Z", "source_reliability": 0.95, "measurement_uncertainty": 0.02 }) # Returns: {"score": 0.92, "category": "good"} ``` ## Advanced Features ### Data Enrichment Automatically add calculated fields and contextual metadata: ```python theme={null} from qubit.adapters.enrichment import DataEnricher enricher = DataEnricher() # Add calculated fields @enricher.calculator("solar_efficiency") def calculate_efficiency(data): if "irradiance" in data.metadata and data.metric == "energy_generation": panel_area = data.asset.specifications.get("panel_area_m2", 1000) irradiance = data.metadata["irradiance_wm2"] theoretical_max = (irradiance * panel_area) / 1000 # kW efficiency = data.value / theoretical_max if theoretical_max > 0 else 0 return {"solar_efficiency": round(efficiency, 3)} return {} # Add weather context @enricher.context_provider("weather") async def add_weather_context(data): if hasattr(data, 'location'): weather = await weather_api.get_current( data.location.latitude, data.location.longitude ) return { "weather_condition": weather.condition, "cloud_cover_percent": weather.cloud_cover } return {} ``` ### Anomaly Detection Built-in anomaly detection for data quality assurance: ```python theme={null} from qubit.adapters.validation import AnomalyDetector detector = AnomalyDetector( metric="energy_generation", model="isolation_forest" ) # Train on historical data training_data = load_historical_generation_data() detector.fit(training_data) # Detect anomalies in real-time @connector.on_message async def detect_anomalies(data): anomaly_score = detector.predict(data.value) if anomaly_score > 0.8: # High anomaly score data.metadata["anomaly_detected"] = True data.metadata["anomaly_score"] = anomaly_score data.quality = "questionable" return data ``` ## Configuration Management ### Environment-based Configuration ```yaml Production theme={null} # config/production.yaml adapters: units: strict_validation: true decimal_precision: 6 timezone: default_timezone: "UTC" validate_transitions: true validation: schema_version: "0.2" quality_threshold: 0.8 enable_anomaly_detection: true logging: level: "INFO" structured: true ``` ```yaml Development theme={null} # config/development.yaml adapters: units: strict_validation: false decimal_precision: 3 timezone: default_timezone: "America/Los_Angeles" validate_transitions: false validation: schema_version: "0.2" quality_threshold: 0.5 enable_anomaly_detection: false logging: level: "DEBUG" structured: false ``` ### Dynamic Configuration ```python theme={null} from qubit.adapters.config import ConfigManager # Load configuration with environment overrides config = ConfigManager.load("production.yaml") # Runtime configuration updates config.update("adapters.validation.quality_threshold", 0.9) # Environment variable overrides # QUBIT_ADAPTERS_UNITS_STRICT_VALIDATION=true ``` ## Performance Optimization For high-throughput applications, use batch processing and connection pooling to maximize performance. ### Batch Processing ```python theme={null} from qubit.adapters import BatchProcessor processor = BatchProcessor( batch_size=1000, max_wait_time=1.0 # seconds ) @processor.batch_handler async def process_batch(data_batch): # Process 1000 records at once validated = validator.validate_batch(data_batch) normalized = unit_converter.convert_batch(validated) # Bulk insert to database await database.insert_many(normalized) # Messages automatically batched await processor.add(timeseries_data) ``` ### Caching ```python theme={null} from qubit.adapters.cache import AdapterCache # Cache expensive operations cache = AdapterCache(ttl=300) # 5-minute TTL @cache.memoize def expensive_validation(schema_name, data_hash): return validator.validate(data) @cache.memoize def timezone_lookup(timezone_name): return pytz.timezone(timezone_name) ``` ## Error Handling Robust error handling ensures data integrity: ```python theme={null} try: normalized_data = adapter.process(raw_data) except UnitConversionError as e: # Log error but continue with original units logger.warning(f"Unit conversion failed: {e}") normalized_data = raw_data normalized_data.quality = "questionable" except ValidationError as e: # Critical error - cannot process logger.error(f"Schema validation failed: {e}") raise ``` ```python theme={null} from qubit.adapters.retry import RetryAdapter @RetryAdapter( max_attempts=3, backoff_factor=2.0, exceptions=[ValidationError, UnitConversionError] ) async def process_with_retry(data): return adapter.process(data) ``` ```python theme={null} from qubit.adapters.queue import DeadLetterQueue dlq = DeadLetterQueue("failed_validations") try: result = adapter.process(data) except Exception as e: # Send to DLQ for manual review await dlq.send(data, error=str(e)) raise ``` ## Testing Always test adapters with representative data from your actual devices before deploying to production. ### Unit Testing ```python theme={null} import pytest from qubit.adapters.units import PowerConverter def test_power_conversion(): # Test standard conversions assert PowerConverter.to_kilowatts(1000, "W") == 1.0 assert PowerConverter.to_kilowatts(2.5, "MW") == 2500.0 # Test edge cases assert PowerConverter.to_kilowatts(0, "W") == 0.0 # Test error handling with pytest.raises(UnitConversionError): PowerConverter.to_kilowatts(100, "invalid_unit") def test_timezone_conversion(): from qubit.adapters.timezone import TimezoneAdapter import datetime adapter = TimezoneAdapter() # Test DST transition local_time = datetime.datetime(2024, 3, 10, 2, 30) # Spring forward utc_time = adapter.to_utc(local_time, "America/Los_Angeles") assert utc_time.hour == 10 # 2:30 AM PST -> 10:30 UTC ``` ### Integration Testing ```python theme={null} from qubit.connectors.mqtt import MQTTConnector from qubit.adapters import AdapterPipeline @pytest.mark.integration async def test_mqtt_to_schema_pipeline(): # Set up test MQTT broker test_broker = await setup_test_broker() # Configure connector with adapters connector = MQTTConnector({ "broker": test_broker.url, "topics": ["test/device/+/readings"] }) pipeline = AdapterPipeline([ ("units", UnitConverter()), ("timezone", TimezoneAdapter()), ("validation", SchemaValidator("timeseries")) ]) # Send test message await test_broker.publish( "test/device/001/readings", {"power": 1500, "unit": "W", "timestamp": "2024-01-15 14:30:00"} ) # Verify normalized output message = await connector.receive() result = pipeline.process(message) assert result.value == 1.5 # Converted to kW assert result.unit == "kW" assert result.timestamp.endswith("Z") # UTC format ``` ## Best Practices ### Performance Guidelines Process data in batches of 100-1000 records for optimal throughput Reuse database connections and validator instances Use async/await for I/O bound operations Stream large datasets instead of loading into memory ### Data Quality Perform validation as close to the data source as possible Keep original values in metadata for debugging and auditing Use quality indicators to flag questionable data for review Track validation failure rates and common error patterns ## Real-World Examples ### Solar Inverter Data ```python theme={null} # Raw inverter data (various units and timezone) raw_data = { "inverter_id": "INV_001", "power_output": 1500, # Watts "energy_today": 45.2, # kWh "timestamp": "2024-01-15 14:30:00", # Local time "timezone": "America/Los_Angeles", "temperature": 77.5, # Fahrenheit "efficiency": 96.2 # Percent } # Process through adapters adapter_pipeline = AdapterPipeline([ PowerConverter(target_unit="kW"), TimezoneConverter(target_timezone="UTC"), TemperatureConverter(target_unit="C"), SchemaValidator("timeseries") ]) result = adapter_pipeline.process(raw_data) # Normalized output { "id": "ts_inv_001_2024_01_15_22_30", "asset_id": "ast_inverter_001", "metric": "power_generation", "value": 1.5, "unit": "kW", "timestamp": "2024-01-15T22:30:00Z", # UTC "quality": "good", "metadata": { "original_value": 1500, "original_unit": "W", "original_timestamp": "2024-01-15 14:30:00 PST", "ambient_temperature_c": 25.3, "inverter_efficiency": 0.962 } } ``` ### EV Charging Data ```python theme={null} # OCPP message processing ocpp_message = { "messageType": "MeterValues", "chargePointId": "CP_001", "meterValue": { "timestamp": "2024-01-15T14:30:00.000Z", "sampledValue": { "value": "15000", # String format "unit": "W", # Watts "measurand": "Power.Active.Import" } } } # Process through OCPP adapter ocpp_adapter = OCPPAdapter() timeseries = ocpp_adapter.process(ocpp_message) # Output { "id": "ts_cp_001_power_2024_01_15_14_30", "asset_id": "ast_charger_cp_001", "metric": "charging_power", "value": 15.0, "unit": "kW", "timestamp": "2024-01-15T14:30:00Z", "quality": "good", "metadata": { "ocpp_measurand": "Power.Active.Import", "charging_session": "active" } } ``` ## Monitoring Adapters Track adapter performance and data quality: ```python theme={null} from qubit.adapters.monitoring import AdapterMetrics metrics = AdapterMetrics() class MonitoredUnitConverter(UnitConverter): def convert(self, value, from_unit, to_unit): start_time = time.time() try: result = super().convert(value, from_unit, to_unit) metrics.record_conversion_success(from_unit, to_unit) return result except Exception as e: metrics.record_conversion_failure(from_unit, to_unit, str(e)) raise finally: duration = time.time() - start_time metrics.record_conversion_latency(duration) # View metrics print(f"Conversion success rate: {metrics.success_rate():.2%}") print(f"Average latency: {metrics.avg_latency():.3f}s") ``` ## Next Steps Ready to start processing energy data? Follow our step-by-step guide. Or explore related components: Understand the data models your adapters will produce Learn how connectors and adapters work together *** *Qubit Adapters ensure that regardless of how your energy data arrives, it's transformed into high-quality, standardized formats ready for analysis and optimization.* # Connectors Source: https://docs.qubit.energy/layer-1/connectors Protocol-specific adapters for ingesting energy data from various sources # Qubit Energy Connectors Protocol-specific connectors for ingesting energy data from various sources into the Qubit Energy platform. These connectors handle the complexity of different communication protocols and translate raw device data into standardized formats. ## Overview Energy systems communicate using hundreds of different protocols. Qubit Connectors provide a unified interface that abstracts away protocol complexity while maintaining full feature support for each protocol's unique capabilities. Real-time IoT device telemetry and messaging EV charger communication (1.6J & 2.0.1) Industrial equipment and smart meter integration Bulk historical data import and REST APIs ## Architecture All connectors follow a consistent architecture pattern: ```mermaid theme={null} graph LR A[Raw Protocol Data] --> B[Protocol Connector] B --> C[Data Adapter] C --> D[Schema Validation] D --> E[TimeSeries Output] subgraph "Connector Layer" B C D end ``` ## Installation ```bash theme={null} # Install base package pip install qubit-energy-connectors # With specific protocol support pip install qubit-energy-connectors[mqtt] pip install qubit-energy-connectors[modbus] pip install qubit-energy-connectors[ocpp] pip install qubit-energy-connectors[all] ``` ```bash theme={null} # Run MQTT connector docker run -d \ --name qubit-mqtt-connector \ -e MQTT_BROKER=mqtt://broker.example.com:1883 \ -e MQTT_TOPICS="+/+/meter/+/readings" \ qubit/energy-connectors:mqtt # Run with configuration file docker run -d \ --name qubit-connectors \ -v $(pwd)/config.json:/app/config.json \ qubit/energy-connectors:latest ``` ```yaml theme={null} apiVersion: apps/v1 kind: Deployment metadata: name: qubit-mqtt-connector spec: replicas: 3 selector: matchLabels: app: qubit-mqtt-connector template: metadata: labels: app: qubit-mqtt-connector spec: containers: - name: connector image: qubit/energy-connectors:mqtt-latest env: - name: MQTT_BROKER value: "mqtt://cluster.example.com:1883" ``` ## MQTT Connector ### Features * **MQTT 3.1.1** and **5.0** support * **QoS levels** 0, 1, and 2 * **TLS/SSL encryption** with certificate validation * **WebSocket transport** for browser compatibility * **Auto-reconnection** with exponential backoff * **Hierarchical topic structures** (`org/site/asset/device/metric`) * **Wildcard subscriptions** (`+` single level, `#` multi-level) * **Dynamic topic discovery** and subscription * **Topic filtering** and routing rules * **JSON message parsing** with schema validation * **Binary payload support** (Protobuf, MessagePack) * **Batch message processing** for high throughput * **Message deduplication** and ordering ### Configuration ```json Basic Configuration theme={null} { "mqtt": { "broker": "mqtt://broker.example.com:1883", "client_id": "qubit_connector_01", "topics": [ "org_001/sit_solar_01/+/+/telemetry", "org_001/sit_storage_01/+/+/telemetry" ], "qos": 1, "keepalive": 60 } } ``` ```json TLS Configuration theme={null} { "mqtt": { "broker": "mqtts://secure-broker.example.com:8883", "client_id": "qubit_connector_secure", "topics": ["+/+/meter/+/readings"], "qos": 2, "tls": { "ca_certs": "/path/to/ca.pem", "certfile": "/path/to/client.crt", "keyfile": "/path/to/client.key", "insecure": false }, "auth": { "username": "${MQTT_USERNAME}", "password": "${MQTT_PASSWORD}" } } } ``` ### Usage Example ```python theme={null} from qubit.connectors.mqtt import MQTTConnector import asyncio # Configuration config = { "broker": "mqtt://localhost:1883", "topics": ["org_001/sit_solar_01/+/+/telemetry"], "username": "user", "password": "pass" } # Data handler async def handle_solar_data(data): """Process incoming solar telemetry.""" print(f"Solar generation: {data['value']} {data['unit']}") # Automatically normalized to TimeSeries schema timeseries = data.to_timeseries() return timeseries # Main application async def main(): connector = MQTTConnector(config) # Connect and subscribe await connector.connect() await connector.subscribe( "org_001/sit_solar_01/+/+/telemetry", handle_solar_data ) # Keep running await asyncio.sleep(3600) # Cleanup await connector.disconnect() asyncio.run(main()) ``` ## OCPP Connector OCPP (Open Charge Point Protocol) is the industry standard for EV charger communication. Qubit supports both OCPP 1.6J and 2.0.1. ### Features * **OCPP 1.6J** and **2.0.1** protocol support * **WebSocket transport** with automatic reconnection * **Message queuing** for offline operation * **Transaction management** with session tracking * **Smart charging** capabilities * **Firmware update management** ### Configuration Example ```python theme={null} from qubit.connectors.ocpp import OCPPConnector config = { "version": "2.0.1", "websocket_url": "wss://chargers.network.com/ocpp", "charge_point_id": "CP_001", "auth": { "basic_auth": { "username": "charger_001", "password": "secure_password" } } } connector = OCPPConnector(config) @connector.on_message("MeterValues") async def handle_meter_values(message): # Convert OCPP message to TimeSeries return TimeSeries( id=f"ts_{message.charge_point_id}_{message.timestamp}", asset_id=message.charge_point_id, metric="charging_power", value=message.meter_value.sampled_value.value, unit="kW", timestamp=message.timestamp ) ``` ## Modbus Connector ### Protocol Support ```python theme={null} config = { "protocol": "rtu", "port": "/dev/ttyUSB0", "baudrate": 19200, "parity": "even", "stopbits": 1, "bytesize": 8, "timeout": 1.0 } ``` ```python theme={null} config = { "protocol": "tcp", "host": "192.168.1.100", "port": 502, "timeout": 5.0, "unit_id": 1 } ``` ### Register Mapping ```json theme={null} { "modbus": { "host": "192.168.1.100", "port": 502, "registers": { "active_power": { "address": 40001, "type": "float32", "scale": 1000, "unit": "kW" }, "energy_total": { "address": 40003, "type": "uint32", "scale": 100, "unit": "kWh" } }, "polling_interval": 5 } } ``` ## CSV Connector ### Bulk Data Import ```python theme={null} from qubit.connectors.csv import CSVConnector config = { "file_path": "/data/historical_generation.csv", "mapping": { "timestamp": "DateTime", "asset_id": "InverterID", "value": "Generation_kWh", "unit": "kWh" }, "timestamp_format": "%Y-%m-%d %H:%M:%S", "timezone": "America/Los_Angeles" } connector = CSVConnector(config) # Process large files with progress tracking async def process_historical_data(): async for timeseries_batch in connector.process(): print(f"Processed {len(timeseries_batch)} records") # Send to Layer 2 for processing await layer2_client.send(timeseries_batch) ``` ## Monitoring and Observability * Connection status and uptime * Message throughput and latency * Error rates and retry counts * Queue depths and processing lag * Schema validation success rates * Data completeness metrics * Outlier detection and flagging * Source reliability scoring ### Example Monitoring ```python theme={null} from qubit.monitoring import ConnectorMetrics metrics = ConnectorMetrics() @connector.on_message async def track_message_metrics(message): metrics.increment_counter("messages_received") metrics.record_histogram("message_size", len(message)) # Process message result = await process_message(message) if result.valid: metrics.increment_counter("messages_valid") else: metrics.increment_counter("messages_invalid") metrics.record_error("validation_failed", result.error) ``` ## Next Steps Select the connector that matches your energy devices Set up authentication and connection parameters Start with a small subset of devices to validate data flow Deploy across your full device fleet with monitoring Ready to connect your first device? Follow our getting started guide. *** *Qubit Connectors handle the complexity of energy protocols so you can focus on building innovative applications on top of clean, standardized data.* # Getting Started Source: https://docs.qubit.energy/layer-1/getting-started Set up your first energy data connector in minutes # Getting Started with Layer 1 Get your energy data flowing through the Qubit Foundation infrastructure in just a few steps. This guide walks you through connecting your first device and processing energy data. ## Prerequisites This guide assumes you have basic familiarity with energy systems and Python programming. For production deployments, consult with your system administrator. * **Python 3.8+** or **Node.js 16+** * **Git** for cloning repositories * **Docker** (optional, for containerized deployment) * **Access to energy device** or historical data * MQTT broker connectivity (port 1883/8883) * OCPP WebSocket endpoints (port 80/443) * Modbus TCP connectivity (port 502) * Outbound HTTPS for schema validation ## Quick Start: 5-Minute Setup ### 1. Install Qubit Components ```bash theme={null} # Install schemas and connectors pip install qubit-energy-schemas qubit-energy-connectors[mqtt] # Clone examples git clone https://github.com/qubit-foundation/qubit-energy-schemas.git cd qubit-energy-schemas ``` ```bash theme={null} # Run pre-configured MQTT connector docker run -d \ --name qubit-mqtt \ -e MQTT_BROKER=mqtt://demo.qubit.energy:1883 \ -e MQTT_TOPICS=demo/solar/+/+/readings \ qubit/connectors:mqtt-latest ``` ```bash theme={null} # Install Node.js packages npm install @qubit/energy-schemas @qubit/connectors # Clone examples git clone https://github.com/qubit-foundation/qubit-energy-schemas.git cd qubit-energy-schemas ``` ### 2. Choose Your Data Source Use our live demo data to get started immediately: ```python theme={null} from qubit.connectors.mqtt import MQTTConnector import asyncio # Demo configuration config = { "broker": "mqtt://demo.qubit.energy:1883", "topics": ["demo/solar/+/+/readings"], "client_id": "demo_client_001" } async def handle_demo_data(data): print(f"Demo data: {data}") return data async def main(): connector = MQTTConnector(config) await connector.connect() await connector.subscribe("demo/solar/+/+/readings", handle_demo_data) await asyncio.sleep(60) # Run for 1 minute await connector.disconnect() asyncio.run(main()) ``` Connect to your existing MQTT infrastructure: ```python theme={null} config = { "broker": "mqtt://your-broker.example.com:1883", "topics": ["your/topic/pattern/+/readings"], "username": "your_username", "password": "your_password", "tls": { "ca_certs": "/path/to/ca.pem" # If using TLS } } ``` Start with historical data to understand the flow: ```python theme={null} from qubit.connectors.csv import CSVConnector config = { "file_path": "historical_generation.csv", "mapping": { "timestamp": "DateTime", "asset_id": "InverterID", "value": "Generation_kWh", "unit": "kWh" }, "timestamp_format": "%Y-%m-%d %H:%M:%S", "timezone": "America/Los_Angeles" } ``` ### 3. Process and Validate Data ```python theme={null} from qubit.adapters import AdapterPipeline from qubit.adapters.units import UnitConverter from qubit.adapters.timezone import TimezoneAdapter from qubit.adapters.validation import SchemaValidator # Create processing pipeline pipeline = AdapterPipeline([ UnitConverter(target_units={"power": "kW", "energy": "kWh"}), TimezoneAdapter(target_timezone="UTC"), SchemaValidator("timeseries", version="0.2") ]) # Process incoming data @connector.on_message async def process_energy_data(raw_message): try: # Run through adapter pipeline timeseries = pipeline.process(raw_message) print(f"✅ Processed: {timeseries.id}") print(f" Asset: {timeseries.asset_id}") print(f" Value: {timeseries.value} {timeseries.unit}") print(f" Quality: {timeseries.quality}") # Ready for Layer 2 processing! return timeseries except Exception as e: print(f"❌ Processing failed: {e}") raise ``` ## Complete Example: Solar Farm Monitoring Let's build a complete solar farm monitoring system: ### 1. Project Structure ``` solar_monitoring/ ├── config/ │ ├── production.yaml │ └── development.yaml ├── src/ │ ├── main.py │ ├── handlers.py │ └── storage.py ├── requirements.txt └── docker-compose.yml ``` ### 2. Configuration ```yaml config/production.yaml theme={null} # Solar farm monitoring configuration mqtt: broker: "mqtts://solar-farm-mqtt.example.com:8883" client_id: "solar_monitor_prod" topics: - "farm/inverter/+/telemetry" - "farm/weather/+/observations" qos: 1 tls: ca_certs: "/certs/ca.pem" certfile: "/certs/client.crt" keyfile: "/certs/client.key" auth: username: "${MQTT_USERNAME}" password: "${MQTT_PASSWORD}" adapters: units: power_target: "kW" energy_target: "kWh" temperature_target: "C" timezone: site_timezone: "America/Los_Angeles" target_timezone: "UTC" validation: schema_version: "0.2" quality_threshold: 0.8 storage: timeseries_db: "postgresql://user:pass@db.example.com/energy" influxdb_url: "http://influx.example.com:8086" ``` ```python requirements.txt theme={null} qubit-energy-schemas>=0.2.0 qubit-energy-connectors[mqtt]>=1.0.0 qubit-energy-adapters>=1.0.0 asyncio pydantic influxdb-client psycopg2-binary ``` ### 3. Implementation ```python src/main.py theme={null} import asyncio import yaml from qubit.connectors.mqtt import MQTTConnector from qubit.adapters import AdapterPipeline from handlers import SolarDataHandler from storage import TimeSeriesStore async def main(): # Load configuration with open('config/production.yaml') as f: config = yaml.safe_load(f) # Set up components connector = MQTTConnector(config['mqtt']) adapter_pipeline = AdapterPipeline.from_config(config['adapters']) storage = TimeSeriesStore(config['storage']) handler = SolarDataHandler(adapter_pipeline, storage) # Connect and start processing await connector.connect() for topic in config['mqtt']['topics']: await connector.subscribe(topic, handler.process_message) print("🌞 Solar farm monitoring started") try: # Run indefinitely while True: await asyncio.sleep(1) except KeyboardInterrupt: print("👋 Shutting down...") finally: await connector.disconnect() await storage.close() if __name__ == "__main__": asyncio.run(main()) ``` ```python src/handlers.py theme={null} from qubit.adapters import AdapterPipeline from qubit.schemas import TimeSeries import logging logger = logging.getLogger(__name__) class SolarDataHandler: def __init__(self, adapter_pipeline, storage): self.pipeline = adapter_pipeline self.storage = storage async def process_message(self, topic: str, message: dict): """Process incoming MQTT message from solar equipment.""" try: # Determine message type from topic topic_parts = topic.split('/') device_type = topic_parts[1] # inverter, weather device_id = topic_parts[2] if device_type == "inverter": return await self._process_inverter_data(device_id, message) elif device_type == "weather": return await self._process_weather_data(device_id, message) except Exception as e: logger.error(f"Failed to process message from {topic}: {e}") raise async def _process_inverter_data(self, device_id: str, data: dict): """Process solar inverter telemetry.""" # Run through adapter pipeline timeseries_list = self.pipeline.process_inverter_batch(data) # Store all metrics for ts in timeseries_list: await self.storage.insert(ts) logger.info(f"Stored {ts.metric}: {ts.value} {ts.unit}") return timeseries_list async def _process_weather_data(self, device_id: str, data: dict): """Process weather station data.""" timeseries = self.pipeline.process_weather(data) await self.storage.insert(timeseries) return timeseries ``` ### 4. Run and Monitor ```bash Development theme={null} # Set environment variables export MQTT_USERNAME=your_username export MQTT_PASSWORD=your_password # Run locally python src/main.py ``` ```bash Production theme={null} # Use Docker Compose docker-compose up -d # Check logs docker-compose logs -f solar-monitor # Monitor metrics curl http://localhost:8080/metrics ``` ```yaml docker-compose.yml theme={null} version: '3.8' services: solar-monitor: build: . environment: - CONFIG_FILE=config/production.yaml - MQTT_USERNAME=${MQTT_USERNAME} - MQTT_PASSWORD=${MQTT_PASSWORD} volumes: - ./config:/app/config - ./certs:/certs depends_on: - influxdb - postgres restart: unless-stopped influxdb: image: influxdb:2.0 environment: - INFLUXDB_DB=energy - INFLUXDB_ADMIN_USER=admin - INFLUXDB_ADMIN_PASSWORD=password volumes: - influx_data:/var/lib/influxdb2 postgres: image: postgres:14 environment: - POSTGRES_DB=energy_metadata - POSTGRES_USER=energy - POSTGRES_PASSWORD=secure_password volumes: - postgres_data:/var/lib/postgresql/data volumes: influx_data: postgres_data: ``` ## Verification Once your system is running, verify data flow: ### 1. Check Data Quality ```python theme={null} # Query recent data from storage import TimeSeriesStore store = TimeSeriesStore(config) recent_data = await store.query( asset_id="ast_inverter_001", metric="power_generation", start_time="2024-01-15T14:00:00Z", end_time="2024-01-15T15:00:00Z" ) # Analyze quality quality_stats = { "good": sum(1 for ts in recent_data if ts.quality == "good"), "questionable": sum(1 for ts in recent_data if ts.quality == "questionable"), "poor": sum(1 for ts in recent_data if ts.quality == "poor") } print(f"Data quality distribution: {quality_stats}") ``` ### 2. Monitor Performance ```python theme={null} # Check connector health health_status = await connector.health_check() print(f"Connector status: {health_status}") # View processing metrics metrics = pipeline.get_metrics() print(f"Messages processed: {metrics['total_processed']}") print(f"Validation success rate: {metrics['validation_success_rate']:.2%}") print(f"Average processing time: {metrics['avg_processing_time']:.3f}s") ``` ## Common Integration Patterns ### Pattern 1: Real-time Dashboard ```python theme={null} # Stream processed data to dashboard @connector.on_message async def stream_to_dashboard(timeseries): if timeseries.metric in ["power_generation", "energy_consumption"]: await dashboard_websocket.send(timeseries.json()) ``` ### Pattern 2: Data Lake Storage ```python theme={null} # Batch write to data lake batch_writer = BatchWriter( destination="s3://energy-data-lake/timeseries/", batch_size=1000, flush_interval=60 # seconds ) @connector.on_message async def store_in_data_lake(timeseries): await batch_writer.add(timeseries) ``` ### Pattern 3: Alert Generation ```python theme={null} # Generate alerts for anomalies from qubit.alerts import AlertManager alert_manager = AlertManager() @connector.on_message async def check_for_alerts(timeseries): if timeseries.quality == "poor": await alert_manager.send_alert( level="warning", message=f"Poor data quality from {timeseries.asset_id}", metadata=timeseries.metadata ) ``` ## Troubleshooting ```python theme={null} # Test MQTT connection from qubit.connectors.mqtt import MQTTConnector connector = MQTTConnector(config) try: await connector.connect(timeout=10) print("✅ Connected successfully") except ConnectionError as e: print(f"❌ Connection failed: {e}") # Check firewall, credentials, broker URL ``` ```python theme={null} # Debug validation failures from qubit.adapters.validation import SchemaValidator validator = SchemaValidator("timeseries", version="0.2") result = validator.validate(data) if not result.is_valid: print("Validation errors:") for error in result.errors: print(f" - {error.path}: {error.message}") ``` ```python theme={null} # Debug unit conversions from qubit.adapters.units import UnitConverter try: converted = UnitConverter.convert(1500, "W", "kW") print(f"Converted: {converted} kW") except UnitConversionError as e: print(f"Conversion error: {e}") print(f"Supported units: {UnitConverter.supported_units()}") ``` ```python theme={null} # Monitor processing performance import time @connector.on_message async def monitor_performance(data): start_time = time.time() result = await pipeline.process(data) processing_time = time.time() - start_time if processing_time > 0.1: # >100ms logger.warning(f"Slow processing: {processing_time:.3f}s") return result ``` ## Production Deployment ### High Availability Setup ```yaml theme={null} apiVersion: apps/v1 kind: Deployment metadata: name: qubit-energy-connector spec: replicas: 3 selector: matchLabels: app: energy-connector template: spec: containers: - name: connector image: qubit/connectors:latest env: - name: CONFIG_FILE value: "/config/production.yaml" resources: requests: cpu: 100m memory: 128Mi limits: cpu: 500m memory: 512Mi livenessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 30 periodSeconds: 10 ``` ```yaml theme={null} version: '3.8' services: energy-connector: image: qubit/connectors:latest deploy: replicas: 3 update_config: parallelism: 1 delay: 10s restart_policy: condition: on-failure environment: - MQTT_BROKER=mqtt://production-broker:1883 volumes: - config:/app/config networks: - energy_network ``` ### Monitoring and Alerting ```python theme={null} # Comprehensive monitoring setup from qubit.monitoring import PrometheusExporter, AlertManager # Export metrics to Prometheus prometheus = PrometheusExporter(port=8080) prometheus.start() # Set up alerts alerts = AlertManager() @connector.on_connect async def on_connection(): await alerts.send("info", "Connector started successfully") @connector.on_disconnect async def on_disconnection(): await alerts.send("warning", "Connector disconnected") @pipeline.on_validation_failure async def on_validation_failure(error, data): if error.severity == "critical": await alerts.send("critical", f"Schema validation failed: {error}") ``` ## Next Steps Learn about production deployment patterns and scaling strategies Add AI-powered forecasting to your energy data Build connectors for proprietary or emerging protocols Set up comprehensive observability and alerting ## Community and Support Connect with other developers in our [GitHub Discussions](https://github.com/qubit-foundation/qubit-energy-schemas/discussions) Found a bug? [Create an issue](https://github.com/qubit-foundation/qubit-energy-connectors/issues) on GitHub Help improve the connectors by [contributing code](https://github.com/qubit-foundation/qubit-energy-connectors/blob/main/CONTRIBUTING.md) *** *You've now established the data foundation for your energy system. Layer 1 provides clean, standardized data ready for advanced processing, optimization, and coordination in the upper layers of the Qubit Foundation stack.* # Schemas Source: https://docs.qubit.energy/layer-1/schemas Universal JSON Schema definitions for energy data interchange # Qubit Energy Schemas [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) [![Schema Version](https://img.shields.io/badge/version-v0.2.0-blue.svg)](https://github.com/qubit-foundation/qubit-energy-schemas/releases) Open-source JSON Schema definitions for energy data interchange. Built by the energy community, for the energy community. ## Vision Creating a universal, open standard for energy data that enables seamless integration between energy assets, monitoring systems, and optimization platforms. Our schemas provide a common language for the energy transition. ## Core Schemas ### Foundation Schemas Energy company, utility, or entity managing energy assets Physical location containing energy assets (solar farm, charging station) Physical energy equipment (solar panels, batteries, inverters) Energy measurement devices providing consumption/generation data Environmental or operational sensors (weather, temperature, vibration) Time-stamped measurements and telemetry data ### Advanced Schemas (v0.2) Dynamic pricing with carbon intensity, P2P trading, V2G rates VPP dispatches, blockchain verification, AI anomaly detection Hyperlocal weather with satellite, drone, and AI-enhanced data ML baselines, V2G coordination, flexibility market participation Predictive maintenance alerts and cybersecurity threat detection ## Quick Start ### Installation ```bash theme={null} # Clone the schemas repository git clone https://github.com/qubit-foundation/qubit-energy-schemas.git cd qubit-energy-schemas # Install validation dependencies pip install -r requirements.txt ``` ### Validate Example Data ```bash theme={null} python scripts/validate.py examples/site.json ``` ### Use in Your Project ```python theme={null} import json from jsonschema import validate # Load schema with open('schemas/v0.2/site.json') as f: schema = json.load(f) # Your data site_data = { "id": "sit_solar_farm_001", "name": "North Solar Farm", "organization_id": "org_acme_energy", "location": { "latitude": 37.7749, "longitude": -122.4194, "timezone": "America/Los_Angeles" } } # Validate validate(instance=site_data, schema=schema) print("✅ Data is valid!") ``` ```javascript theme={null} const Ajv = require('ajv'); const fs = require('fs'); // Load schema const schema = JSON.parse(fs.readFileSync('schemas/v0.2/timeseries.json')); const ajv = new Ajv(); const validate = ajv.compile(schema); // Your data const timeseriesData = { "id": "ts_solar_001_2024_01_15_14_30", "asset_id": "ast_solar_inverter_001", "metric": "energy_generation", "value": 2500.0, "unit": "kWh", "timestamp": "2024-01-15T14:30:00Z" }; // Validate const valid = validate(timeseriesData); if (!valid) console.log(validate.errors); ``` ```go theme={null} package main import ( "encoding/json" "github.com/xeipuuv/gojsonschema" ) func validateTimeSeries(data []byte) error { schemaLoader := gojsonschema.NewReferenceLoader("file:///schemas/v0.2/timeseries.json") documentLoader := gojsonschema.NewBytesLoader(data) result, err := gojsonschema.Validate(schemaLoader, documentLoader) if err != nil { return err } if !result.Valid() { return fmt.Errorf("validation failed: %v", result.Errors()) } return nil } ``` ## Schema Examples ### TimeSeries Schema The most commonly used schema for real-time energy data: ```json theme={null} { "id": "ts_solar_farm_001_2024_01_15_14_30", "asset_id": "ast_solar_farm_001", "metric": "energy_generation", "value": 2500.0, "unit": "kWh", "timestamp": "2024-01-15T14:30:00Z", "quality": "good", "metadata": { "weather_condition": "sunny", "panel_temperature": 35.2, "inverter_efficiency": 0.98 } } ``` ### Asset Schema Defines physical energy equipment: ```json theme={null} { "id": "ast_battery_storage_001", "name": "Tesla Megapack 2XL", "site_id": "sit_energy_storage_facility", "type": "battery_storage", "specifications": { "capacity_kwh": 3916, "max_charge_rate_kw": 1890, "max_discharge_rate_kw": 1890, "efficiency": 0.89, "chemistry": "lithium_iron_phosphate" }, "installation": { "commissioned_date": "2024-01-15", "warranty_years": 20, "installer": "Tesla Energy" } } ``` ## Key Features * **Organizations**: `org_` prefix (e.g., `org_pacific_gas_electric`) * **Sites**: `sit_` prefix (e.g., `sit_solar_farm_001`) * **Assets**: `ast_` prefix (e.g., `ast_inverter_001`) * **Meters**: `mtr_` prefix (e.g., `mtr_main_001`) * **Sensors**: `sns_` prefix (e.g., `sns_temperature_001`) * **TimeSeries**: `ts_` prefix (e.g., `ts_generation_001`) All timestamps use ISO 8601 UTC format: `2024-01-15T14:30:00Z` Standardized on International System of Units (kWh, kW, °C, m/s, etc.) * Smart contract addresses and transaction hashes * Cryptographic verification fields * Token-based settlement integration * Predictive maintenance indicators * Anomaly detection confidence scores * ML model metadata and versioning * Dynamic tariff structures * Carbon intensity tracking * Grid congestion pricing * Quality indicators: `good`, `questionable`, `poor` * Confidence scores for AI-generated data * Data source provenance tracking * `metadata` objects for custom fields * Backward-compatible versioning * Plugin architecture for domain-specific extensions * Comprehensive JSON Schema validation * Automated testing with 1000+ example records * CI/CD integration with validation workflows ## Schema Versions Always specify which schema version you're using. Version 0.2 includes significant enhancements for blockchain, AI, and advanced energy markets. ### Version 0.2 (Current) * Advanced tariff structures with dynamic pricing * Blockchain and smart contract integration * AI/ML metadata and confidence scoring * V2G and bidirectional energy flow support * Enhanced cybersecurity and threat detection ### Version 0.1 (Legacy) * Foundation schemas for basic energy data * Simple tariff structures * Core asset and measurement definitions ## Validation Tools ```bash Validate Single File theme={null} python scripts/validate.py examples/timeseries.json ``` ```bash Validate All Examples theme={null} python scripts/validate.py examples/ ``` ```bash Generate Types theme={null} python scripts/generate_types.py --output types/ ``` ## Contributing Fork the [schemas repository](https://github.com/qubit-foundation/qubit-energy-schemas) ```bash theme={null} git checkout -b feature/new-schema-enhancement ``` * Add or modify schema files in `schemas/v0.2/` * Include example data in `examples/` * Update documentation ```bash theme={null} python scripts/validate.py pytest tests/ ``` Create a pull request with detailed description of changes ## Best Practices Always validate your data against the schemas before sending to production systems. Invalid data can cause downstream processing failures. ### Schema Design Guidelines 1. **Minimize Breaking Changes**: Use optional fields for new features 2. **Clear Naming**: Use descriptive, unambiguous field names 3. **Consistent Units**: Always specify units explicitly 4. **Rich Metadata**: Include context and provenance information 5. **Future-Proof**: Design for extensibility and evolution ### Implementation Tips Cache compiled schemas for repeated validation to improve throughput Provide meaningful error messages when validation fails Track validation success rates and common error patterns Support multiple schema versions during migration periods ## Real-World Examples ### Solar Farm Monitoring ```json theme={null} { "id": "ts_solar_generation_2024_01_15_14_30", "asset_id": "ast_solar_array_001", "metric": "energy_generation", "value": 1250.5, "unit": "kWh", "timestamp": "2024-01-15T14:30:00Z", "quality": "good", "metadata": { "irradiance_wm2": 850, "ambient_temp_c": 25.3, "panel_temp_c": 45.1, "inverter_efficiency": 0.976 } } ``` ### EV Charging Session ```json theme={null} { "id": "ts_charging_power_2024_01_15_15_45", "asset_id": "ast_fast_charger_001", "metric": "charging_power", "value": 150.0, "unit": "kW", "timestamp": "2024-01-15T15:45:00Z", "quality": "good", "metadata": { "session_id": "ses_charge_001", "vehicle_id": "veh_tesla_model_s_001", "soc_percent": 65, "charging_curve": "fast" } } ``` *** *The Qubit Energy Schemas provide the data foundation that enables interoperability across the entire energy ecosystem. By establishing universal standards, we accelerate the development and deployment of innovative energy solutions.* # Forecasting Models Source: https://docs.qubit.energy/layer-2/forecasting Deep dive into AI-powered energy forecasting models and algorithms # Energy Forecasting Models Layer 2 provides state-of-the-art machine learning models specifically designed for energy domain challenges. Each model is optimized for different forecasting horizons, data patterns, and accuracy requirements. ## Model Categories Classical and modern approaches for temporal pattern recognition Renewable generation forecasting with meteorological data Demand forecasting incorporating human behavior patterns Combining multiple models for optimal accuracy ## Available Models ### Classical Time Series **Auto-Regressive Integrated Moving Average** Best for: Stable patterns, linear trends, short-term forecasting ```python theme={null} from qubit.forecasting.models import ARIMAForecaster forecaster = ARIMAForecaster( order=(2, 1, 2), # (p, d, q) seasonal_order=(1, 1, 1, 24), # Daily seasonality auto_select=True # Automatic parameter selection ) forecaster.fit(historical_data) forecast = forecaster.predict(horizon="24h") ``` **Strengths:** * Fast training and inference * Interpretable parameters * Confidence intervals included * Good for stable seasonal patterns **Limitations:** * Assumes linear relationships * Limited external feature support * Poor with non-stationary data **Facebook Prophet** Best for: Seasonal data, holiday effects, trend changes ```python theme={null} from qubit.forecasting.models import ProphetForecaster forecaster = ProphetForecaster( yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=True, holidays=['US'], # Built-in holiday calendar changepoint_prior_scale=0.05 ) # Automatic feature engineering forecaster.fit(energy_data) forecast = forecaster.predict( horizon="7d", include_history=False ) ``` **Strengths:** * Handles missing data gracefully * Automatic holiday detection * Robust to outliers * Interpretable trend/seasonality decomposition **Best Use Cases:** * Load forecasting with strong seasonal patterns * Long-term capacity planning * Data with irregular patterns **Holt-Winters and ETS Models** Best for: Simple seasonality, fast deployment ```python theme={null} from qubit.forecasting.models import ExponentialSmoothingForecaster forecaster = ExponentialSmoothingForecaster( trend='add', seasonal='add', seasonal_periods=24, # Hourly seasonality damped_trend=True ) forecaster.fit(load_history) forecast = forecaster.predict(horizon="48h") ``` ### Machine Learning Models **Ensemble of Decision Trees** Best for: Feature-rich data, non-linear patterns, uncertainty quantification ```python theme={null} from qubit.forecasting.models import RandomForestForecaster forecaster = RandomForestForecaster( n_estimators=100, max_depth=15, min_samples_split=5, bootstrap=True ) # Include weather and calendar features features = feature_engineer.extract( energy_data, include_weather=True, include_calendar=True, include_lag=True ) forecaster.fit(features, target_values) forecast = forecaster.predict(future_features) ``` **Feature Importance Analysis:** ```python theme={null} importance = forecaster.get_feature_importance() print(importance.head(10)) # Output: # temperature 0.234 # hour_sin 0.187 # load_lag_24h 0.156 # is_weekend 0.098 # cloud_cover 0.089 ``` **Gradient Boosting** Best for: High accuracy, complex feature interactions ```python theme={null} from qubit.forecasting.models import XGBoostForecaster forecaster = XGBoostForecaster( n_estimators=200, max_depth=8, learning_rate=0.1, subsample=0.8, colsample_bytree=0.8 ) # Advanced hyperparameter tuning forecaster.tune_hyperparameters( X_train, y_train, cv_folds=5, n_trials=100 ) ``` **Quantile Regression:** ```python theme={null} # Uncertainty quantification forecast = forecaster.predict( X_future, quantiles=[0.1, 0.25, 0.5, 0.75, 0.9] ) ``` **SVR with RBF Kernel** Best for: Non-linear patterns, small datasets ```python theme={null} from qubit.forecasting.models import SVRForecaster forecaster = SVRForecaster( kernel='rbf', C=1.0, gamma='scale', epsilon=0.1 ) ``` ### Deep Learning Models **Long Short-Term Memory** Best for: Long-term dependencies, complex temporal patterns ```python theme={null} from qubit.forecasting.models import LSTMForecaster forecaster = LSTMForecaster( lstm_units=[128, 64], dropout=0.2, lookback_window=168, # 1 week of hourly data forecast_horizon=24, batch_size=32, epochs=100 ) # Multi-step ahead prediction forecaster.fit( X_train, y_train, validation_split=0.2, early_stopping=True ) forecast = forecaster.predict(X_test) ``` **Architecture Visualization:** ```python theme={null} forecaster.plot_model_architecture() forecaster.plot_training_history() ``` **Sequence-to-Sequence Prediction:** ```python theme={null} # Input: 7 days of hourly data # Output: Next 24 hours forecast = forecaster.predict_sequence( input_sequence, horizon=24 ) ``` **Gated Recurrent Units** Best for: Faster training than LSTM, good performance ```python theme={null} from qubit.forecasting.models import GRUForecaster forecaster = GRUForecaster( gru_units=[64, 32], dropout=0.3, recurrent_dropout=0.2, return_sequences=True ) ``` **Attention-Based Architecture** Best for: Very long sequences, parallel training ```python theme={null} from qubit.forecasting.models import TransformerForecaster forecaster = TransformerForecaster( d_model=64, n_heads=8, n_layers=4, d_ff=256, max_seq_length=8760 # Full year ) # Self-attention visualization attention_weights = forecaster.get_attention_weights(input_data) forecaster.plot_attention_heatmap(attention_weights) ``` ## Domain-Specific Forecasters ### Solar Generation Forecasting Combines physics-based solar calculations with ML corrections: ```python theme={null} from qubit.forecasting.solar import PhysicalSolarForecaster forecaster = PhysicalSolarForecaster( system_capacity_kw=1000, panel_type='monocrystalline', tilt_angle=35, azimuth=180, location={'lat': 37.7749, 'lon': -122.4194} ) # Uses solar position algorithms + weather data forecast = forecaster.predict( weather_forecast, include_physics=True, ml_correction=True ) ``` Satellite imagery analysis for sub-hour forecasting: ```python theme={null} from qubit.forecasting.solar import CloudMotionForecaster forecaster = CloudMotionForecaster( satellite_provider='goes_16', tracking_method='optical_flow', nowcasting_horizon='3h' ) # Requires satellite imagery API forecast = forecaster.predict( current_irradiance, satellite_images=imagery_data ) ``` System-specific performance modeling: ```python theme={null} from qubit.forecasting.solar import PVPerformanceForecaster forecaster = PVPerformanceForecaster( pv_model='sandia', inverter_model='cec', degradation_rate=0.005 # Annual ) # Temperature coefficient correction forecast = forecaster.predict( irradiance_forecast, temperature_forecast, wind_forecast ) ``` ### Demand & Load Forecasting Different models for different customer types: ```python theme={null} from qubit.forecasting.demand import SegmentedDemandForecaster forecaster = SegmentedDemandForecaster( segments={ 'residential': ResidentialModel(), 'commercial': CommercialModel(), 'industrial': IndustrialModel() } ) # Automatic customer classification forecast = forecaster.predict( customer_data, auto_segment=True ) ``` Incorporates electricity pricing effects: ```python theme={null} from qubit.forecasting.demand import PriceElasticDemandForecaster forecaster = PriceElasticDemandForecaster( price_elasticity=-0.1, # 10% demand reduction per 100% price increase income_elasticity=0.3 ) forecast = forecaster.predict( historical_load, price_forecast=prices_df ) ``` Temperature-dependent demand modeling: ```python theme={null} from qubit.forecasting.demand import WeatherSensitiveForecaster forecaster = WeatherSensitiveForecaster( cooling_threshold=22, # Celsius heating_threshold=18, weather_lag=1 # 1-hour temperature lag ) ``` ### Wind Generation Forecasting Physics-based wind-to-power conversion: ```python theme={null} from qubit.forecasting.wind import WindPowerForecaster forecaster = WindPowerForecaster( turbine_type='vestas_v90', hub_height=80, # meters rated_power=2000, # kW cut_in_speed=3, # m/s cut_out_speed=25 # m/s ) # Wind speed to power conversion forecast = forecaster.predict( wind_speed_forecast, wind_direction_forecast ) ``` Turbine interaction effects in wind farms: ```python theme={null} from qubit.forecasting.wind import WindFarmForecaster forecaster = WindFarmForecaster( turbine_layout=turbine_coordinates, wake_model='jensen', terrain_roughness=0.1 ) ``` ## Ensemble Methods ### Model Combination Strategies ```python theme={null} from qubit.forecasting.ensemble import WeightedEnsemble ensemble = WeightedEnsemble([ ('prophet', ProphetForecaster()), ('xgboost', XGBoostForecaster()), ('lstm', LSTMForecaster()) ], weights=[0.3, 0.4, 0.3]) # Automatic weight optimization ensemble.optimize_weights( X_train, y_train, method='minimize_mse' ) ``` ```python theme={null} from qubit.forecasting.ensemble import StackingEnsemble base_models = [ ('arima', ARIMAForecaster()), ('rf', RandomForestForecaster()), ('svr', SVRForecaster()) ] ensemble = StackingEnsemble( base_models=base_models, meta_learner=XGBoostForecaster() ) ensemble.fit(X_train, y_train) forecast = ensemble.predict(X_test) ``` ```python theme={null} from qubit.forecasting.ensemble import DynamicEnsemble ensemble = DynamicEnsemble([ ('short_term', ARIMAForecaster()), ('medium_term', ProphetForecaster()), ('long_term', LSTMForecaster()) ]) # Different models for different horizons forecast = ensemble.predict( X_test, horizon_weights={ '1h': [0.7, 0.2, 0.1], '24h': [0.3, 0.5, 0.2], '7d': [0.1, 0.3, 0.6] } ) ``` ## Model Selection Guide Choose models based on your specific use case, data characteristics, and performance requirements. ### Decision Matrix | Use Case | Recommended Model | Alternative | Training Time | Inference Speed | | -------------------------- | ----------------- | --------------------- | ------------- | --------------- | | **Short-term load (1-4h)** | XGBoost | ARIMA | Medium | Fast | | **Day-ahead solar** | Ensemble | Random Forest | Slow | Medium | | **Week-ahead demand** | Prophet | LSTM | Fast | Fast | | **Real-time pricing** | ARIMA | SVR | Fast | Very Fast | | **Seasonal planning** | Prophet | Exponential Smoothing | Fast | Fast | | **Complex patterns** | LSTM | Transformer | Very Slow | Slow | ### Data Requirements * ARIMA * Exponential Smoothing * SVR * Simple ensemble * Random Forest * XGBoost * Prophet * Weighted ensemble * LSTM/GRU * Transformer * Deep ensemble * Stacking ensemble * Distributed XGBoost * Multi-GPU LSTM * Transformer with attention * Neural ensemble ## Advanced Features ### Uncertainty Quantification All models support multiple uncertainty estimation methods: ```python theme={null} # Prediction intervals forecast = forecaster.predict( X_test, confidence_levels=[0.5, 0.8, 0.95], method='quantile_regression' ) # Monte Carlo dropout (for neural networks) forecast = lstm_forecaster.predict( X_test, uncertainty_method='mc_dropout', mc_samples=100 ) # Ensemble variance forecast = ensemble.predict( X_test, return_std=True, return_individual_predictions=True ) ``` ### Online Learning Models that adapt to new data automatically: ```python theme={null} from qubit.forecasting.online import OnlineForecaster online_model = OnlineForecaster( base_model=XGBoostForecaster(), update_frequency='1h', window_size=8760, # 1 year rolling window adaptation_rate=0.01 ) # Continuous learning for new_data_point in data_stream: prediction = online_model.predict(new_data_point.features) online_model.update(new_data_point.features, new_data_point.target) ``` ### Multi-Horizon Forecasting Generate predictions for multiple time horizons simultaneously: ```python theme={null} from qubit.forecasting.multi import MultiHorizonForecaster forecaster = MultiHorizonForecaster( horizons=['1h', '6h', '24h', '7d'], models={ '1h': ARIMAForecaster(), '6h': XGBoostForecaster(), '24h': LSTMForecaster(), '7d': ProphetForecaster() } ) multi_forecast = forecaster.predict(X_test) print(f"1-hour: {multi_forecast['1h'].peak_value:.2f} kW") print(f"24-hour: {multi_forecast['24h'].total:.2f} kWh") ``` ## Model Evaluation ### Comprehensive Metrics ```python theme={null} from qubit.forecasting.evaluation import ForecastEvaluator evaluator = ForecastEvaluator( metrics=['mape', 'rmse', 'mae', 'peak_accuracy', 'energy_score'] ) results = evaluator.evaluate( y_true=test_targets, forecasts=model_predictions, timestamps=test_timestamps ) # Energy-specific metrics print(f"MAPE: {results.mape:.2%}") print(f"Peak timing accuracy: {results.peak_accuracy:.2%}") print(f"Energy score: {results.energy_score:.3f}") ``` ### Cross-Validation Time series aware cross-validation: ```python theme={null} from qubit.forecasting.validation import TimeSeriesCrossValidator cv = TimeSeriesCrossValidator( n_splits=5, gap=24, # 24-hour gap between train/test horizon=24 # 24-hour forecast horizon ) cv_scores = cv.cross_validate( forecaster=XGBoostForecaster(), X=features, y=targets, metrics=['mape', 'rmse'] ) print(f"CV MAPE: {cv_scores['mape'].mean():.2%} ± {cv_scores['mape'].std():.2%}") ``` ## Next Steps Build your first forecast in 5 minutes Explore the full source code and examples *** *The forecasting models are continuously improved based on real-world deployment feedback and cutting-edge ML research.* # Getting Started Source: https://docs.qubit.energy/layer-2/getting-started Install and run your first energy forecast in 5 minutes # Getting Started with Layer 2 Forecasting Get energy predictions running in just a few minutes with Qubit's AI-powered forecasting models. This guide walks you through installation, setup, and generating your first forecasts. ## Installation ```bash theme={null} pip install qubit-energy-forecasting # Or with specific ML backends pip install qubit-energy-forecasting[all] # All models pip install qubit-energy-forecasting[deep] # TensorFlow/Keras pip install qubit-energy-forecasting[prophet] # Facebook Prophet ``` ```python theme={null} import qubit.forecasting print(f"Qubit Forecasting v{qubit.forecasting.__version__}") # Check available models from qubit.forecasting import SolarForecaster, DemandForecaster print("✅ Installation successful!") ``` ## Quick Examples ### Solar Generation Forecast Generate a 24-hour solar generation forecast: ```python Solar Forecast Example theme={null} from qubit.forecasting import SolarForecaster import pandas as pd import numpy as np # 1. Create sample historical data dates = pd.date_range('2024-01-01', periods=720, freq='1H') np.random.seed(42) # Simulate solar generation pattern hours = dates.hour seasonal_factor = 0.8 + 0.4 * np.sin(2 * np.pi * dates.dayofyear / 365) daily_pattern = np.where( (hours >= 6) & (hours <= 18), 1000 * np.exp(-((hours - 12) ** 2) / 18), 0 ) historical_data = pd.DataFrame({ 'irradiance': daily_pattern * seasonal_factor + np.random.normal(0, 50, len(dates)), 'temperature': 25 + np.random.normal(0, 5, len(dates)), 'cloud_cover': np.random.beta(2, 5, len(dates)), 'generation': daily_pattern * seasonal_factor * 0.18 + np.random.normal(0, 10, len(dates)) }, index=dates) historical_data = historical_data.clip(lower=0) # 2. Initialize and train forecaster forecaster = SolarForecaster( system_capacity_kw=1000, panel_type="polycrystalline", tilt_angle=35 ) # Prepare features and target X = historical_data[['irradiance', 'temperature', 'cloud_cover']] y = historical_data['generation'] # Train the model forecaster.fit(X, y) print("✅ Model trained successfully!") # 3. Generate forecast # Create future weather forecast (normally from weather API) future_weather = pd.DataFrame({ 'irradiance': [800, 850, 900, 950, 1000, 950, 900, 800, 600, 300, 0, 0, 0, 0, 0, 0, 0, 0, 200, 600, 850, 950, 900, 800], 'temperature': [20, 22, 25, 27, 30, 32, 30, 28, 25, 22, 18, 15, 12, 10, 10, 12, 15, 18, 20, 22, 25, 27, 25, 22], 'cloud_cover': [0.2] * 24 }, index=pd.date_range(dates[-1] + pd.Timedelta(hours=1), periods=24, freq='1H')) forecast = forecaster.predict( X.tail(24), # Last 24 hours for context horizon="24h", weather_forecast=future_weather, confidence_levels=[0.80, 0.95], asset_id="solar_farm_001" ) print(f"🌞 24-hour Solar Forecast Generated!") print(f"Peak generation: {forecast.peak_value:.2f} kW at {forecast.peak_time.strftime('%H:%M')}") print(f"Total energy: {forecast.total:.2f} kWh") print(f"Confidence: 80% interval [{forecast.confidence_intervals[0.8][0].max():.0f}, {forecast.confidence_intervals[0.8][1].max():.0f}] kW") ``` ```python View Results theme={null} # 4. Analyze the forecast forecast_df = forecast.to_dataframe() print("\n📊 Hourly Forecast:") print(forecast_df[['timestamp', 'forecast', 'lower_80', 'upper_80']].head(12)) # Feature importance importance = forecaster.get_feature_importance() print("\n🔍 Top Features:") print(importance.head(5)) # Export to Qubit TimeSeries format timeseries_data = forecast.to_timeseries() print(f"\n💾 Exported {len(timeseries_data)} TimeSeries records") print("Sample record:", timeseries_data[0]) ``` ```python Plot Results (Optional) theme={null} import matplotlib.pyplot as plt # Plot the forecast fig, ax = plt.subplots(figsize=(12, 6)) ax.plot(forecast_df['timestamp'], forecast_df['forecast'], label='Forecast', linewidth=2, color='red') ax.fill_between(forecast_df['timestamp'], forecast_df['lower_80'], forecast_df['upper_80'], alpha=0.3, color='red', label='80% Confidence') ax.set_title('24-Hour Solar Generation Forecast') ax.set_xlabel('Time') ax.set_ylabel('Generation (kW)') ax.legend() ax.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() plt.savefig('solar_forecast.png') plt.show() ``` ### Demand Forecast Generate electricity demand predictions: ```python Demand Forecast Example theme={null} from qubit.forecasting import DemandForecaster import pandas as pd import numpy as np # 1. Create sample load data dates = pd.date_range('2024-01-01', periods=720, freq='1H') np.random.seed(42) # Simulate demand patterns hours = dates.hour weekday_pattern = 50 + 30 * ( ((hours >= 7) & (hours <= 9)) | # Morning peak ((hours >= 17) & (hours <= 21)) # Evening peak ).astype(int) weekend_pattern = 40 + 10 * np.sin(2 * np.pi * hours / 24) is_weekend = dates.dayofweek >= 5 base_load = np.where(is_weekend, weekend_pattern, weekday_pattern) temperature_effect = 2 * (25 - np.abs(dates.hour - 12)) # Temperature effect historical_load = pd.DataFrame({ 'temperature': 20 + 10 * np.sin(2 * np.pi * dates.dayofyear / 365) + np.random.normal(0, 2, len(dates)), 'load': base_load + temperature_effect + np.random.normal(0, 5, len(dates)) }, index=dates) historical_load['load'] = historical_load['load'].clip(lower=0) # 2. Initialize and train demand forecaster demand_forecaster = DemandForecaster( customer_type="mixed", include_weather=True, include_calendar=True ) # Prepare data X_demand = historical_load[['temperature']] y_demand = historical_load['load'] # Train model demand_forecaster.fit(X_demand, y_demand) print("✅ Demand model trained successfully!") # 3. Generate demand forecast future_temp = pd.DataFrame({ 'temperature': 22 + 3 * np.sin(2 * np.pi * np.arange(24) / 24) + np.random.normal(0, 1, 24) }, index=pd.date_range(dates[-1] + pd.Timedelta(hours=1), periods=24, freq='1H')) demand_forecast = demand_forecaster.predict( X_demand.tail(24), horizon="24h", temperature_forecast=future_temp, asset_id="load_mixed_001" ) print(f"⚡ 24-hour Demand Forecast Generated!") print(f"Peak demand: {demand_forecast.peak_value:.2f} kW at {demand_forecast.peak_time.strftime('%H:%M')}") print(f"Average demand: {demand_forecast.point_forecast.mean():.2f} kW") # Peak metrics peak_metrics = demand_forecaster.calculate_peak_metrics(demand_forecast) print(f"Load factor: {peak_metrics['load_factor']:.2%}") ``` ## Integration with Layer 1 Layer 2 is designed to work seamlessly with Layer 1 data: ```python Layer 1 Integration theme={null} from qubit.connectors.mqtt import MQTTConnector from qubit.forecasting import SolarForecaster from qubit.adapters import AdapterPipeline # 1. Set up Layer 1 data pipeline connector = MQTTConnector({ "broker": "mqtt://your-broker.com:1883", "topics": ["solar/+/telemetry"] }) adapter_pipeline = AdapterPipeline([ ("units", UnitConverter()), ("timezone", TimezoneAdapter()), ("validation", SchemaValidator("timeseries")) ]) # 2. Initialize forecaster forecaster = SolarForecaster(system_capacity_kw=1000) # 3. Process real-time data and forecast @connector.on_message async def process_and_forecast(mqtt_message): # Layer 1: Normalize data timeseries = adapter_pipeline.process(mqtt_message) # Layer 2: Generate forecast when enough data if len(historical_buffer) > 168: # 1 week of data forecast = forecaster.predict( historical_buffer.tail(24), horizon="24h" ) # Send forecast to Layer 3 or storage await send_to_layer3(forecast.to_timeseries()) await connector.start() ``` ```python Batch Processing theme={null} from qubit.forecasting import SolarForecaster import pandas as pd # Process historical TimeSeries data def batch_forecast(timeseries_file: str): # Load Layer 1 TimeSeries data df = pd.read_csv(timeseries_file) # Filter for solar generation data solar_data = df[df['metric'] == 'energy_generation'] # Group by asset forecasts = [] for asset_id, asset_data in solar_data.groupby('asset_id'): forecaster = SolarForecaster(system_capacity_kw=1000) # Prepare features from TimeSeries data features = pd.DataFrame({ 'generation': asset_data['value'].values }, index=pd.to_datetime(asset_data['timestamp'])) # Simple forecast without weather data if len(features) > 168: # Need minimum data forecast = forecaster.predict( features.tail(24), horizon="24h", asset_id=asset_id ) forecasts.append(forecast) return forecasts # Process batch forecasts = batch_forecast('energy_timeseries.csv') print(f"Generated {len(forecasts)} forecasts") ``` ## Configuration ### Environment Variables ```bash theme={null} # Optional configuration export QUBIT_FORECASTING_CACHE_DIR="/tmp/qubit_models" export QUBIT_FORECASTING_LOG_LEVEL="INFO" export QUBIT_FORECASTING_PARALLEL_JOBS="4" ``` ### Configuration File ```yaml config.yaml theme={null} forecasting: default_horizon: "24h" default_resolution: "1h" cache_models: true models: solar: default_capacity: 1000 # kW include_weather: true confidence_levels: [0.8, 0.95] demand: customer_segments: ["residential", "commercial", "industrial"] include_calendar: true performance: batch_size: 32 n_jobs: -1 memory_limit: "4GB" ``` ```python Load Configuration theme={null} from qubit.forecasting.config import load_config config = load_config('config.yaml') forecaster = SolarForecaster.from_config(config['models']['solar']) ``` ## Docker Deployment Run forecasting models in containers: ```dockerfile Dockerfile theme={null} FROM python:3.9-slim # Install forecasting package RUN pip install qubit-energy-forecasting[all] # Copy your forecasting script COPY forecast_app.py /app/ WORKDIR /app # Run the forecasting service CMD ["python", "forecast_app.py"] ``` ```yaml docker-compose.yml theme={null} version: '3.8' services: solar-forecaster: build: . environment: - FORECASTING_MODEL=solar - HORIZON=24h - RESOLUTION=1h volumes: - ./data:/app/data - ./models:/app/models ports: - "8080:8080" demand-forecaster: build: . environment: - FORECASTING_MODEL=demand - CUSTOMER_TYPE=mixed volumes: - ./data:/app/data ``` ```bash Deploy theme={null} # Build and run docker-compose up -d # Check status docker-compose ps # View logs docker-compose logs solar-forecaster ``` ## API Server Deploy forecasting as a REST API: ```python api_server.py theme={null} from fastapi import FastAPI, HTTPException from pydantic import BaseModel from qubit.forecasting import SolarForecaster, DemandForecaster import pandas as pd app = FastAPI(title="Qubit Energy Forecasting API") # Initialize models solar_forecaster = SolarForecaster(system_capacity_kw=1000) demand_forecaster = DemandForecaster(customer_type="mixed") class ForecastRequest(BaseModel): asset_id: str model_type: str # "solar" or "demand" horizon: str = "24h" historical_data: dict weather_data: dict = None @app.post("/forecast") async def generate_forecast(request: ForecastRequest): try: if request.model_type == "solar": forecaster = solar_forecaster elif request.model_type == "demand": forecaster = demand_forecaster else: raise HTTPException(400, "Invalid model_type") # Convert data to DataFrame X = pd.DataFrame(request.historical_data) # Generate forecast forecast = forecaster.predict( X, horizon=request.horizon, asset_id=request.asset_id ) return { "asset_id": forecast.asset_id, "forecast": forecast.point_forecast.tolist(), "timestamps": forecast.timestamps.isoformat().tolist(), "peak_value": forecast.peak_value, "total": forecast.total } except Exception as e: raise HTTPException(500, str(e)) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080) ``` ```bash Start API Server theme={null} # Install dependencies pip install fastapi uvicorn # Run server python api_server.py # Test API curl -X POST "http://localhost:8080/forecast" \ -H "Content-Type: application/json" \ -d '{ "asset_id": "solar_001", "model_type": "solar", "horizon": "24h", "historical_data": {"generation": [100, 200, 300]} }' ``` ## Troubleshooting ```bash theme={null} # Missing dependencies pip install --upgrade qubit-energy-forecasting[all] # Python version issues python --version # Requires 3.9+ # Virtual environment python -m venv venv source venv/bin/activate # Linux/Mac venv\Scripts\activate # Windows pip install qubit-energy-forecasting ``` ```python theme={null} # Reduce batch size forecaster = XGBoostForecaster( batch_size=16, # Reduce from default 32 n_jobs=2 # Limit parallel jobs ) # Clear model cache forecaster.clear_cache() # Use lighter models forecaster = ARIMAForecaster() # Instead of LSTM ``` ```python theme={null} # Enable model caching forecaster.cache_model = True # Pre-compute features features = feature_engineer.extract(data) forecaster.fit(features, target, cache_features=True) # Use ensemble weights ensemble.optimize_weights(X_train, y_train, method='fast') ``` ```python theme={null} # Validate input data from qubit.forecasting.validation import validate_timeseries errors = validate_timeseries(data) if errors: print("Data validation errors:", errors) # Handle missing values data_cleaned = data.fillna(method='ffill').fillna(method='bfill') # Remove outliers from qubit.forecasting.preprocessing import remove_outliers data_clean = remove_outliers(data, method='iqr', threshold=3) ``` ## Next Steps Learn about different forecasting algorithms and when to use them Deploy forecasting models at scale with Kubernetes and monitoring Browse complete example applications and use cases Use forecasts for energy system optimization *** *You're now ready to generate production-quality energy forecasts! Layer 2 provides the predictions that power intelligent energy optimization in Layer 3.* # Layer 2 Overview Source: https://docs.qubit.energy/layer-2/overview AI-powered forecasting and prediction engine for energy systems # Layer 2: Prediction Engine [![Production Ready](https://img.shields.io/badge/status-production-green)](https://github.com/qubit-foundation/qubit-energy-forecasting) [![Python](https://img.shields.io/badge/python-3.9%2B-blue)](https://www.python.org/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) Layer 2 provides production-ready machine learning models for energy forecasting, demand prediction, and renewable generation analysis. Building on standardized data from Layer 1, this layer delivers sub-second predictions for energy system optimization. ## Available Features Weather-based solar PV generation predictions with confidence intervals Electricity demand prediction with behavioral and seasonal patterns Automated feature extraction for energy time series data Sub-second inference with batch and real-time prediction modes ## Quick Start ### Installation ```bash theme={null} pip install qubit-energy-forecasting # With specific ML backends pip install qubit-energy-forecasting[deep] # TensorFlow pip install qubit-energy-forecasting[prophet] # Facebook Prophet pip install qubit-energy-forecasting[all] # All models ``` ### Generate Your First Forecast ```python theme={null} from qubit.forecasting import SolarForecaster import pandas as pd # Initialize forecaster forecaster = SolarForecaster( system_capacity_kw=1000, panel_type="polycrystalline" ) # Train on historical data forecaster.fit(X_train, y_train) # Generate 24-hour forecast forecast = forecaster.predict( weather_data, horizon="24h", confidence_levels=[0.80, 0.95] ) print(f"Peak generation: {forecast.peak_value:.2f} kW") print(f"Total energy: {forecast.total:.2f} kWh") ``` ## Architecture ```mermaid theme={null} graph LR A[Layer 1 Data] --> B[Feature Engineering] B --> C[ML Models] C --> D[Predictions] D --> E[Layer 3 Optimization] subgraph "ML Pipeline" B C end subgraph "Models" F[Solar Forecaster] G[Demand Forecaster] H[Wind Forecaster] I[Price Forecaster] end C --> F C --> G C --> H C --> I ``` ## Technology Stack * **Python 3.9+** - Primary language * **scikit-learn** - Classical ML algorithms * **TensorFlow/Keras** - Deep learning models * **XGBoost/LightGBM** - Gradient boosting * **Facebook Prophet** - Time series forecasting * **pandas/numpy** - Data manipulation * **Qubit Schemas** - Standardized data formats * **Feature Store** - Automated feature engineering * **Validation** - Schema and range validation * **FastAPI** - REST API serving * **Docker** - Containerization * **Kubernetes** - Orchestration * **Prometheus** - Monitoring and metrics ## Model Performance Performance metrics from production deployments across 100+ energy assets. ### Solar Generation Forecasting | Horizon | MAPE | RMSE | Peak Accuracy | | -------- | ----- | ------- | ------------- | | 1 hour | 8.5% | 12.3 kW | 94.2% | | 24 hours | 15.2% | 28.1 kW | 87.6% | | 7 days | 22.8% | 41.7 kW | 78.3% | ### Demand Forecasting | Customer Type | MAPE | RMSE | Load Factor | | ------------- | ----- | ------- | ----------- | | Residential | 12.1% | 5.8 kW | 0.68 | | Commercial | 8.9% | 15.2 kW | 0.74 | | Industrial | 6.3% | 42.1 kW | 0.82 | ## Real-World Applications * **15-minute** load forecasting for grid balancing * **Day-ahead** renewable integration planning * **Week-ahead** maintenance scheduling optimization * **Seasonal** capacity planning and resource allocation * **Intraday** generation forecasting for trading * **Weather-dependent** O\&M scheduling * **Performance** monitoring and anomaly detection * **Financial** revenue and P\&L forecasting * **Peer-to-peer** trading optimization * **Storage** dispatch scheduling * **EV charging** load coordination * **Demand response** event planning ## Integration with Qubit Stack Layer 2 seamlessly integrates with other Qubit Foundation components: Consumes standardized TimeSeries from **Layer 1** (Schemas, Connectors, Adapters) Automatically extracts time, weather, calendar, and lag features Trains specialized models for each energy domain and asset type Outputs forecasts in TimeSeries format for **Layer 3** optimization ## Available Models ### Time Series Models Classical autoregressive models for stable patterns Handles seasonality and holidays automatically Deep learning for complex temporal dependencies Gradient boosting for feature-rich predictions Ensemble methods with uncertainty quantification Combines multiple models for best accuracy ### Domain-Specific Forecasters ```python theme={null} from qubit.forecasting.solar import SolarForecaster forecaster = SolarForecaster( system_capacity_kw=5000, panel_type="monocrystalline", tilt_angle=35, azimuth=180 ) # Integrates weather data automatically forecast = forecaster.predict( historical_data, weather_forecast=weather_df, horizon="24h" ) ``` ```python theme={null} from qubit.forecasting.demand import DemandForecaster forecaster = DemandForecaster( customer_type="commercial", include_weather=True, include_calendar=True ) forecast = forecaster.predict( load_history, temperature_forecast=temp_df, horizon="24h" ) ``` ## Deployment Options Single-node deployment for development and small installations Production deployment with auto-scaling and high availability Event-driven predictions for variable workloads Local inference for latency-sensitive applications ### Example Deployment ```yaml theme={null} apiVersion: apps/v1 kind: Deployment metadata: name: energy-forecaster spec: replicas: 3 template: spec: containers: - name: forecaster image: qubit/energy-forecasting:latest resources: requests: cpu: 1000m memory: 2Gi env: - name: MODEL_CACHE_SIZE value: "10" ``` ## Next Steps Install and run your first forecast in 5 minutes Deep dive into available prediction models Explore source code and contribute to development Connect Layer 2 with your Layer 1 data pipeline *** *Layer 2 Prediction Engine is production-ready and powering forecasts across renewable energy, utilities, and smart grid applications worldwide.* # EV Charging Optimizer Source: https://docs.qubit.energy/layer-3/ev-scheduler Fleet charging schedule optimization with tariff-aware, priority-based scheduling # EV Charging Optimizer The `EVChargingScheduler` determines the optimal charging power for each vehicle at each time slot, minimizing total energy cost while respecting site capacity, charger limits, vehicle availability, and priority ordering. ## How It Works ```mermaid theme={null} graph TD A[Charging Sessions] --> D[Build Availability Matrix] B[Tariff Data] --> E[Compute Effective Cost] C[Site Constraints] --> F[Greedy LP Solver] D --> F E --> F F --> G[Optimal Schedule] G --> H[Per-Vehicle kW Profile] G --> I[Cost Breakdown] G --> J[Constraint Report] ``` The optimizer uses a **greedy LP-relaxation** approach: 1. **Availability matrix** — For each session × time slot, determine if the vehicle is present 2. **Effective cost** — Combine energy price and carbon intensity with configurable weights 3. **Priority ordering** — Process urgent vehicles first, then by earliest departure 4. **Slot filling** — For each vehicle, fill cheapest available slots until energy need is met 5. **Capacity tracking** — Deduct allocated power from remaining site headroom ## Configuration ```python theme={null} from optimizer.base import OptimizationConfig from optimizer.ev.scheduler import EVChargingScheduler config = OptimizationConfig( horizon="24h", # Scheduling window resolution="1h", # Time slot granularity (supports 15min, 1h) objective="minimize_cost", weights={ "cost": 1.0, # Energy cost weight "carbon": 0.0, # Carbon intensity weight "peak": 0.0 # Peak reduction weight } ) scheduler = EVChargingScheduler( config=config, site_capacity_kw=200.0, # Max total site power draw num_chargers=10, # Number of charging ports charger_capacity_kw=22.0 # Max power per charger (AC Level 2) ) ``` ## Charging Sessions Each EV is described by a `ChargingSession`: ```python theme={null} from optimizer.ev.scheduler import ChargingSession from datetime import datetime, timezone session = ChargingSession( vehicle_id="ev_001", arrival_time=datetime(2025, 1, 15, 8, 0, tzinfo=timezone.utc), departure_time=datetime(2025, 1, 15, 17, 0, tzinfo=timezone.utc), energy_needed_kwh=30.0, # Energy to deliver max_charge_rate_kw=22.0, # Vehicle onboard charger limit min_charge_rate_kw=0.0, # Minimum (0 = can pause) current_soc_percent=20.0, # Current battery level target_soc_percent=80.0, # Desired departure SOC battery_capacity_kwh=60.0, # Vehicle battery size priority=1, # 1=normal, 2=high, 3=urgent v2g_enabled=False, # Vehicle-to-grid capable max_discharge_rate_kw=0.0 # V2G discharge limit ) ``` ### Priority Levels | Priority | Description | Scheduling Behavior | | -------- | ----------- | ------------------------------------------------- | | 1 | Normal | Scheduled after higher priorities, cost-optimized | | 2 | High | Preferred slot allocation over normal priority | | 3 | Urgent | First access to cheapest available slots | ## Tariff Integration The scheduler natively consumes Qubit tariff schemas: ```python theme={null} tariff = { "energy_rates": [ { "name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"} }, { "name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"} }, { "name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"} } ] } ``` Tariff schedules support: * **Time-of-use windows** — `start_time` / `end_time` in HH:MM format * **Weekday filtering** — `weekdays` array (e.g., `["monday", "tuesday"]`) * **Monthly filtering** — `months` array (e.g., `[6, 7, 8]` for summer) * **Carbon intensity** — `carbon_intensity_gco2_kwh` per rate period ## Running an Optimization ```python theme={null} import numpy as np # Multiple vehicles sessions = [ ChargingSession(vehicle_id="ev_001", arrival_hour=8, departure_hour=17, energy_needed_kwh=30.0, max_charge_rate_kw=22.0, priority=1), ChargingSession(vehicle_id="ev_002", arrival_hour=9, departure_hour=15, energy_needed_kwh=20.0, max_charge_rate_kw=22.0, priority=2), ChargingSession(vehicle_id="ev_003", arrival_hour=10, departure_hour=18, energy_needed_kwh=40.0, max_charge_rate_kw=22.0, priority=3), ] # Optional: background site load and solar generation load_forecast = np.full(24, 100.0) # 100 kW constant background load solar_forecast = np.zeros(24) solar_forecast[8:17] = 50.0 # 50 kW solar during day result = scheduler.optimize( sessions=sessions, tariff=tariff, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc), load_forecast=load_forecast, solar_forecast=solar_forecast ) ``` ## Result Structure The `OptimizationResult` contains: ```python theme={null} # Status result.status # "optimal" or "feasible" result.is_optimal # True if all constraints met result.solve_time_ms # Solve time in milliseconds # Metrics result.total_cost # Total energy cost ($) result.total_energy_kwh # Total energy delivered (kWh) result.peak_demand_kw # Maximum net load (kW) result.carbon_kg # Total carbon emissions (kg CO2) # Constraints result.constraints_satisfied # {"energy_ev_001": True, "energy_ev_002": True, "site_capacity": True} # Schedule DataFrame result.schedule.columns # ["ev_ev_001_kw", "ev_ev_002_kw", "ev_ev_003_kw", # "total_ev_kw", "background_load_kw", "solar_generation_kw", # "net_load_kw", "price_per_kwh", "slot_cost"] ``` ### Schedule Columns | Column | Description | | --------------------- | ------------------------------------------------- | | `ev_{vehicle_id}_kw` | Charging power allocated to each vehicle per slot | | `total_ev_kw` | Sum of all EV charging power per slot | | `background_load_kw` | Site background load (from forecast) | | `solar_generation_kw` | Solar generation (from forecast) | | `net_load_kw` | Total net load: background + EV - solar | | `price_per_kwh` | Energy price at each slot | | `slot_cost` | Cost for each time slot | ## Constraint Enforcement The scheduler enforces several hard constraints: Total EV charging power plus background load minus solar never exceeds the site transformer rating. The optimizer tracks remaining headroom at each slot and caps allocation accordingly. Each vehicle's charge rate is capped at the minimum of its onboard charger limit and the EVSE port capacity (`charger_capacity_kw`). Charging only occurs during the arrival-to-departure window. No power is allocated outside these bounds. The optimizer attempts to deliver the full `energy_needed_kwh` for each session. If constraints prevent this (e.g., too many vehicles competing for limited capacity), the result status changes from "optimal" to "feasible". ## Carbon-Aware Scheduling Enable carbon-conscious scheduling by adjusting objective weights: ```python theme={null} config = OptimizationConfig( weights={"cost": 0.5, "carbon": 0.5, "peak": 0.0} ) tariff = { "energy_rates": [ {"name": "night_clean", "rate": 0.08, "carbon_intensity_gco2_kwh": 200, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "day_dirty", "rate": 0.25, "carbon_intensity_gco2_kwh": 450, "schedule": {"start_time": "07:00", "end_time": "22:59"}} ] } ``` The effective cost becomes: `0.5 * price + 0.5 * carbon_intensity`, shifting charging toward lower-carbon periods. ## Next Steps Combine EV scheduling with battery dispatch for demand management Full installation and quickstart guide # Getting Started Source: https://docs.qubit.energy/layer-3/getting-started Install and run your first energy optimization in 5 minutes # Getting Started with Layer 3 Optimization Get cost-optimal energy scheduling running in minutes. This guide walks you through installation, EV charging optimization, and peak shaving with battery storage. ## Installation ```bash theme={null} pip install qubit-energy-optimizer ``` ```python theme={null} from optimizer.base import OptimizationConfig, OptimizationResult from optimizer.ev.scheduler import EVChargingScheduler from optimizer.peak_shaving.controller import PeakShavingController print("Qubit Energy Optimizer installed successfully!") ``` ## Quick Examples ### EV Charging Schedule Optimize charging for a fleet of EVs on a time-of-use tariff: ```python EV Charging Example theme={null} from optimizer.ev.scheduler import EVChargingScheduler, ChargingSession from optimizer.base import OptimizationConfig from datetime import datetime, timezone import numpy as np # 1. Configure the scheduler config = OptimizationConfig(horizon="24h", resolution="1h") scheduler = EVChargingScheduler( config=config, site_capacity_kw=200.0, # Max site power charger_capacity_kw=22.0 # AC Level 2 chargers ) # 2. Define charging sessions sessions = [ ChargingSession( vehicle_id="ev_001", arrival_time=datetime(2025, 1, 15, 8, 0, tzinfo=timezone.utc), departure_time=datetime(2025, 1, 15, 17, 0, tzinfo=timezone.utc), energy_needed_kwh=30.0, max_charge_rate_kw=22.0, priority=1 ), ChargingSession( vehicle_id="ev_002", arrival_time=datetime(2025, 1, 15, 9, 0, tzinfo=timezone.utc), departure_time=datetime(2025, 1, 15, 15, 0, tzinfo=timezone.utc), energy_needed_kwh=20.0, max_charge_rate_kw=22.0, priority=2 # High priority — gets cheaper slots first ), ] # 3. Define tariff tariff = { "energy_rates": [ {"name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"}}, {"name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"}} ] } # 4. Run optimization result = scheduler.optimize( sessions=sessions, tariff=tariff, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc) ) print(f"Status: {result.status}") print(f"Total cost: ${result.total_cost:.2f}") print(f"Total energy: {result.total_energy_kwh:.1f} kWh") print(f"Peak demand: {result.peak_demand_kw:.1f} kW") print(f"Solve time: {result.solve_time_ms:.1f} ms") ``` ```python View Schedule theme={null} # 5. Inspect the schedule schedule = result.schedule print("\nHourly Charging Schedule (kW):") print(schedule[["ev_ev_001_kw", "ev_ev_002_kw", "total_ev_kw", "price_per_kwh"]].to_string()) # Check constraint satisfaction print("\nConstraints:") for name, met in result.constraints_satisfied.items(): print(f" {name}: {'PASS' if met else 'FAIL'}") ``` ### Peak Shaving with Battery Reduce demand peaks using battery storage: ```python Peak Shaving Example theme={null} from optimizer.peak_shaving.controller import PeakShavingController, BatterySpec from optimizer.base import OptimizationConfig from datetime import datetime, timezone import numpy as np # 1. Configure the controller config = OptimizationConfig(horizon="24h", resolution="1h") controller = PeakShavingController( config=config, peak_target_kw=150.0 # Target: keep demand below 150 kW ) # 2. Define battery battery = BatterySpec( asset_id="ast_batt_001", capacity_kwh=200.0, max_charge_kw=50.0, max_discharge_kw=50.0, efficiency_charge=0.95, efficiency_discharge=0.95, min_soc_percent=10.0, max_soc_percent=90.0, initial_soc_percent=50.0 ) # 3. Create load profile with peaks hours = np.arange(24) load = 80 + 100 * np.exp(-((hours - 9) ** 2) / 4) + 120 * np.exp(-((hours - 18) ** 2) / 4) # 4. Define tariff with demand charges tariff = { "energy_rates": [ {"name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"}}, {"name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"}} ], "demand_charges": [ {"name": "monthly_demand", "rate_per_kw": 15.0} ] } # 5. Run optimization result = controller.optimize( battery=battery, load_forecast=load, tariff=tariff, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc) ) print(f"Status: {result.status}") print(f"Peak before: {load.max():.1f} kW") print(f"Peak after: {result.peak_demand_kw:.1f} kW") print(f"Peak reduction: {result.constraints_satisfied['peak_reduction_kw']:.1f} kW") print(f"Demand charge savings: ${result.constraints_satisfied['demand_charge_savings']:.2f}") print(f"Battery wear cost: ${result.constraints_satisfied['battery_wear_cost']:.2f}") ``` ```python View Battery Profile theme={null} # 6. Inspect battery dispatch schedule = result.schedule print("\nBattery Dispatch Schedule:") cols = ["load_forecast_kw", "battery_charge_kw", "battery_discharge_kw", "battery_soc_percent", "net_load_after_kw"] print(schedule[cols].to_string(float_format="%.1f")) ``` ## Integration with Layer 2 Forecasts Use Layer 2 predictions as optimization inputs: ```python theme={null} from forecasting.solar import SolarForecaster from forecasting.demand import DemandForecaster from optimizer.peak_shaving.controller import PeakShavingController, BatterySpec # Generate forecasts (Layer 2) solar_forecast = solar_forecaster.predict(weather_data, horizon="24h") load_forecast = demand_forecaster.predict(features, horizon="24h") # Optimize battery dispatch (Layer 3) result = controller.optimize( battery=battery, load_forecast=load_forecast.point_forecast.values, solar_forecast=solar_forecast.point_forecast.values, tariff=tariff, start_time=start_time ) ``` ## Configuration Reference ### OptimizationConfig | Parameter | Default | Description | | ---------------- | ------------------------------------------- | ------------------------------------------- | | `horizon` | `"24h"` | Optimization window (`"24h"`, `"7d"`, etc.) | | `resolution` | `"15min"` | Time slot size (`"15min"`, `"1h"`) | | `objective` | `"minimize_cost"` | Primary objective function | | `max_iterations` | `1000` | Solver iteration limit | | `tolerance` | `1e-6` | Convergence tolerance | | `weights` | `{"cost": 1.0, "carbon": 0.0, "peak": 0.0}` | Multi-objective weights | ### EVChargingScheduler | Parameter | Default | Description | | --------------------- | ------- | ----------------------------- | | `site_capacity_kw` | `200.0` | Maximum total site power draw | | `num_chargers` | `10` | Number of EVSE ports | | `charger_capacity_kw` | `22.0` | Max power per charging port | ### PeakShavingController | Parameter | Default | Description | | ---------------- | ------- | ----------------------------------------- | | `peak_target_kw` | `None` | Target peak (None = auto 80th percentile) | ## Next Steps Priority scheduling, carbon-aware optimization, and constraint details Two-pass algorithm, SOC management, and demand charge savings Generate the forecasts that feed into Layer 3 Source code, examples, and issue tracker *** *You're now ready to optimize energy schedules! Layer 3 takes forecasts from Layer 2 and tariff data from Layer 1 to produce cost-optimal dispatch decisions.* # Layer 3 Overview Source: https://docs.qubit.energy/layer-3/overview Mathematical optimization engine for intelligent energy resource scheduling and control # Layer 3: Optimization Engine [![Production Ready](https://img.shields.io/badge/status-production-green)](https://github.com/qubit-foundation/qubit-energy-optimizer) [![Python](https://img.shields.io/badge/python-3.9%2B-blue)](https://www.python.org/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) Layer 3 provides mathematical optimization algorithms that determine optimal resource allocation, scheduling, and control strategies for energy systems. Building on forecasts from Layer 2 and tariff data from Layer 1, this layer delivers cost-optimal dispatch schedules in milliseconds. ## Available Features Fleet charging schedule optimization with tariff-aware, priority-based, carbon-conscious scheduling Battery dispatch optimization to minimize peak demand and demand charges using a two-pass algorithm Configurable objective weights for cost, carbon, and peak reduction trade-offs Native consumption of Qubit tariff schemas — time-of-use, demand charges, and carbon intensity ## Quick Start ### Installation ```bash theme={null} pip install qubit-energy-optimizer ``` ### Optimize an EV Charging Schedule ```python theme={null} from optimizer.ev.scheduler import EVChargingScheduler, ChargingSession from optimizer.base import OptimizationConfig from datetime import datetime, timezone config = OptimizationConfig(horizon="24h", resolution="1h") scheduler = EVChargingScheduler( config=config, site_capacity_kw=200.0, charger_capacity_kw=22.0 ) session = ChargingSession( vehicle_id="ev_001", arrival_time=datetime(2025, 1, 15, 8, 0, tzinfo=timezone.utc), departure_time=datetime(2025, 1, 15, 17, 0, tzinfo=timezone.utc), energy_needed_kwh=30.0, max_charge_rate_kw=22.0, priority=1 ) tariff = { "energy_rates": [ {"name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"}}, {"name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"}} ] } result = scheduler.optimize(sessions=[session], tariff=tariff) print(f"Total cost: ${result.total_cost:.2f}") print(f"Peak demand: {result.peak_demand_kw:.1f} kW") ``` ## Architecture ```mermaid theme={null} graph LR A[Layer 1 Tariff Data] --> D[Optimization Engine] B[Layer 2 Load Forecast] --> D C[Layer 2 Solar Forecast] --> D D --> E[Dispatch Schedule] E --> F[Layer 4 Coordination] subgraph "Optimizers" G[EV Charging Scheduler] H[Peak Shaving Controller] end D --> G D --> H ``` ## Technology Stack * **Python 3.9+** — Primary language * **NumPy / pandas** — Numerical computation and schedule DataFrames * **Pydantic** — Configuration validation * **Greedy LP** — Cost-sorted slot filling for EV scheduling * **Two-pass dispatch** — Peak identification + cheapest-slot charging * **Qubit Tariff Schema** — Native TOU, demand charge, and carbon intensity parsing * **Layer 2 Forecasts** — Load and solar forecast arrays as direct inputs * **Layer 1 IDs** — Asset IDs (`ast_*`) carried through optimization results * **Site capacity limits** — Total EV load never exceeds transformer rating * **Charger capacity** — Per-port max charge rate enforcement * **Battery SOC bounds** — Min/max state-of-charge with efficiency modeling * **Vehicle availability** — Charge only during arrival-to-departure windows ## Optimization Algorithms ### EV Charging — Greedy LP The EV scheduler uses a greedy linear-programming relaxation: 1. Build an **availability matrix** — which vehicles are present at each time slot 2. Compute **effective cost** per slot — `cost_weight * price + carbon_weight * carbon_intensity` 3. Process sessions by **priority** (urgent first), then **deadline** (earliest departure first) 4. For each session, fill cheapest available slots until energy requirement is met 5. Enforce **site capacity** by tracking remaining headroom at each slot ### Peak Shaving — Two-Pass Dispatch The peak shaving controller uses a two-pass algorithm: 1. **Pass 1 (Discharge):** Walk forward through time. At each slot where load exceeds the peak target, discharge the battery (respecting SOC limits and max discharge rate) 2. **Pass 2 (Charge):** Calculate total energy discharged, then schedule recharging in the cheapest off-peak slots that have headroom below the peak target This approach naturally produces schedules that: * Shave peak demand below the target threshold * Minimize charging cost by exploiting TOU tariff structure * Respect all battery physical constraints (SOC, charge/discharge rates, efficiency) * Account for battery degradation cost ## Integration with Qubit Stack Consumes **tariff schemas** from Layer 1 and **forecast arrays** from Layer 2 Runs cost-optimal scheduling algorithms with configurable objectives and constraints Produces time-indexed DataFrames with per-slot dispatch commands, SOC profiles, and cost breakdowns Outputs feed into **Layer 4** for real-time dispatch and multi-site coordination ## Next Steps Install and run your first optimization in 5 minutes Deep dive into fleet charging optimization Deep dive into battery dispatch for demand reduction Explore source code and contribute *** *Layer 3 Optimization Engine is production-ready, powering EV charging fleets and battery dispatch across commercial and industrial energy sites.* # Peak Shaving Controller Source: https://docs.qubit.energy/layer-3/peak-shaving Battery dispatch optimization to minimize peak demand and demand charges # Peak Shaving Controller The `PeakShavingController` optimizes battery storage dispatch to reduce peak electricity demand, minimizing demand charges and energy costs. It uses a two-pass algorithm — discharge during peaks, recharge in the cheapest off-peak slots. ## How It Works ```mermaid theme={null} graph TD A[Load Forecast] --> D[Identify Peaks] B[Tariff Data] --> E[Sort Slots by Price] C[Battery Spec] --> F[Two-Pass Dispatch] D --> F E --> F F --> G[Discharge Schedule] F --> H[Charge Schedule] G --> I[Final SOC Profile] H --> I I --> J[Optimized Net Load] ``` ### Two-Pass Algorithm **Pass 1 — Discharge:** Walk forward through each time slot. Wherever the net load exceeds the peak target, discharge the battery to bring it down. Respects SOC minimum, max discharge rate, and round-trip efficiency. **Pass 2 — Charge:** Total up the energy discharged in Pass 1. Schedule recharging in the cheapest available slots (sorted by tariff price) that have headroom below the peak target. This avoids creating new peaks during charging. ## Battery Specification ```python theme={null} from optimizer.peak_shaving.controller import BatterySpec battery = BatterySpec( asset_id="ast_batt_001", capacity_kwh=200.0, # Total battery capacity max_charge_kw=50.0, # Maximum charging power max_discharge_kw=50.0, # Maximum discharging power efficiency_charge=0.95, # One-way charging efficiency efficiency_discharge=0.95, # One-way discharging efficiency min_soc_percent=10.0, # Minimum allowed SOC max_soc_percent=90.0, # Maximum allowed SOC initial_soc_percent=50.0, # Starting SOC degradation_cost_per_kwh=0.02 # Battery wear cost per kWh cycled ) ``` ### Battery Parameters | Parameter | Description | Typical Range | | -------------------------------------------- | ------------------------- | ------------- | | `capacity_kwh` | Nameplate energy capacity | 50–500 kWh | | `max_charge_kw` / `max_discharge_kw` | C-rate power limits | 0.5C–2C | | `efficiency_charge` / `efficiency_discharge` | One-way efficiency | 0.90–0.98 | | `min_soc_percent` / `max_soc_percent` | Usable SOC window | 10–90% | | `degradation_cost_per_kwh` | Wear cost per kWh cycled | \$0.01–0.05 | ## Configuration ```python theme={null} from optimizer.base import OptimizationConfig from optimizer.peak_shaving.controller import PeakShavingController config = OptimizationConfig( horizon="24h", resolution="1h" ) controller = PeakShavingController( config=config, peak_target_kw=150.0 # Target peak demand threshold ) ``` ### Auto-Target Mode If `peak_target_kw=None`, the controller automatically sets the target to the 80th percentile of the forecast load, shaving the top 20% of peak demand: ```python theme={null} controller = PeakShavingController(config=config, peak_target_kw=None) # Target auto-calculated from load forecast ``` ## Running an Optimization ```python theme={null} import numpy as np from datetime import datetime, timezone # Load profile with morning and evening peaks hours = np.arange(24) load = 80 + 100 * np.exp(-((hours - 9) ** 2) / 4) + 120 * np.exp(-((hours - 18) ** 2) / 4) # Optional: solar generation forecast solar = np.zeros(24) solar[8:17] = 60.0 # 60 kW solar during day # TOU tariff with demand charges tariff = { "energy_rates": [ {"name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"}}, {"name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"}} ], "demand_charges": [ {"name": "monthly_demand", "rate_per_kw": 15.0, "measurement_period": "monthly"} ] } result = controller.optimize( battery=battery, load_forecast=load, tariff=tariff, solar_forecast=solar, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc) ) ``` ## Result Structure ```python theme={null} # Status result.status # "optimal" or "feasible" result.peak_demand_kw # Peak net load after optimization result.total_cost # Energy cost + battery wear cost # Savings breakdown result.constraints_satisfied["peak_reduction_kw"] # kW reduced result.constraints_satisfied["cost_savings"] # $ saved on energy result.constraints_satisfied["demand_charge_savings"] # $ saved on demand charges result.constraints_satisfied["battery_wear_cost"] # $ battery degradation ``` ### Schedule Columns | Column | Description | | ---------------------- | ------------------------------------------ | | `load_forecast_kw` | Original load forecast | | `solar_forecast_kw` | Solar generation forecast | | `net_load_before_kw` | Net load before battery (load - solar) | | `battery_charge_kw` | Battery charging power (from grid) | | `battery_discharge_kw` | Battery discharging power (to site) | | `battery_soc_kwh` | Battery state of charge in kWh | | `battery_soc_percent` | Battery state of charge as percentage | | `net_load_after_kw` | Final net load after battery dispatch | | `peak_target_kw` | Peak target threshold | | `price_per_kwh` | Energy price per slot | | `slot_cost_before` | Energy cost per slot (before optimization) | | `slot_cost_after` | Energy cost per slot (after optimization) | ## Constraint Enforcement The battery SOC never drops below `min_soc_percent` or exceeds `max_soc_percent`. Both charge and discharge are clipped to stay within the usable energy window, accounting for round-trip efficiency losses. Power is capped at `max_charge_kw` and `max_discharge_kw` at every time slot. The two-pass algorithm naturally respects these limits during both discharge (Pass 1) and charge (Pass 2) phases. The controller aims to keep net load at or below the target. If the battery is too small to fully shave the peak, the result status is "feasible" rather than "optimal", and the peak is reduced as much as possible. During Pass 2 (charging), the algorithm only allocates power up to the headroom between current load and the peak target. This prevents charging from creating new demand peaks. ## Solar Integration When a solar forecast is provided, the controller works with **net load** (load - solar), which: * Reduces the effective peak that needs shaving * Creates more low-cost charging opportunities during solar hours * Naturally coordinates battery and solar for maximum benefit ```python theme={null} result = controller.optimize( battery=battery, load_forecast=load, solar_forecast=solar, # Reduces net load during solar hours start_time=start_time ) # Net load before battery includes solar offset midday_net = result.schedule["net_load_before_kw"].iloc[12] # load - solar ``` ## Demand Charge Savings The controller calculates demand charge savings when `demand_charges` are present in the tariff: ``` demand_charge_savings = (peak_before - peak_after) * rate_per_kw ``` For example, reducing peak from 200 kW to 150 kW with a $15/kW demand charge saves $750/month. ## Next Steps Combine peak shaving with EV fleet scheduling Full installation and quickstart guide # Dispatch Engine Source: https://docs.qubit.energy/layer-4/dispatch Converting optimization schedules into timed asset commands with lifecycle tracking # Dispatch Engine The `DispatchEngine` is the core bridge between Layer 3's optimization output and Layer 4's command execution. It reads schedule DataFrames and produces a queue of timed `Command` objects, then manages their lifecycle through dispatch, response handling, retries, and timeout detection. ## Schedule-to-Command Conversion The dispatch engine parses Layer 3 schedule DataFrame columns using naming conventions: | Column Pattern | Command Type | Asset Type | | ---------------------- | -------------------- | ---------- | | `ev_{vehicle_id}_kw` | `SET_CHARGE_RATE` | EV Charger | | `battery_charge_kw` | `SET_CHARGE_RATE` | Battery | | `battery_discharge_kw` | `SET_DISCHARGE_RATE` | Battery | ```python theme={null} from coordinator import DispatchEngine, EventBus engine = DispatchEngine(event_bus=EventBus()) # From Layer 3 EV scheduler result commands = engine.schedule_to_commands( result.schedule, asset_map={ "ev_001": "ast_charger_bay_1", "ev_002": "ast_charger_bay_2", "battery": "ast_batt_001" } ) ``` ### Change Detection The engine only emits commands when power values change between consecutive time slots. This avoids flooding assets with redundant setpoints: ``` Hour 8: 22 kW → Command: SET_CHARGE_RATE 22 kW Hour 9: 22 kW → (skipped — same value) Hour 10: 22 kW → (skipped — same value) Hour 11: 0 kW → Command: SET_CHARGE_RATE 0 kW ``` ## Command Lifecycle ```mermaid theme={null} stateDiagram-v2 [*] --> PENDING : schedule_to_commands() PENDING --> DISPATCHED : dispatch_command() DISPATCHED --> COMPLETED : handle_response(success=True) DISPATCHED --> PENDING : handle_response(success=False, retries left) DISPATCHED --> FAILED : handle_response(success=False, retries exhausted) DISPATCHED --> TIMED_OUT : check_timeouts() ``` ### Dispatching Commands ```python theme={null} from datetime import datetime, timezone # Get commands ready for execution now = datetime.now(timezone.utc) pending = engine.get_pending_commands(now) # Dispatch through protocol adapters for cmd in pending: adapter = registry.get(cmd.asset_id) engine.dispatch_command(cmd, adapter) ``` ### Handling Responses ```python theme={null} # Asset acknowledged and executed engine.handle_response(cmd, success=True) # Asset reported an error — will retry if under max_retries engine.handle_response(cmd, success=False, error="connection timeout") ``` ### Timeout Detection ```python theme={null} # Check for commands stuck in DISPATCHED state timed_out = engine.check_timeouts(now) for cmd in timed_out: print(f"Command {cmd.command_id} to {cmd.asset_id} timed out") ``` ## Command Object ```python theme={null} from coordinator.base import Command, CommandType, CommandStatus cmd = Command( asset_id="ast_charger_001", command_type=CommandType.SET_CHARGE_RATE, parameters={"power_kw": 22.0}, scheduled_at=datetime(2025, 1, 15, 8, 0, tzinfo=timezone.utc), # Auto-generated: # command_id="cmd_a1b2c3d4e5f6" # status=CommandStatus.PENDING # max_retries=3 # timeout_seconds=30.0 ) ``` ### Command Types | Type | Description | Typical Target | | --------------------- | ---------------------- | ------------------- | | `SET_CHARGE_RATE` | Set charging power | EV charger, battery | | `SET_DISCHARGE_RATE` | Set discharging power | Battery | | `SET_POWER_LIMIT` | Set power limit | Any | | `START_CHARGING` | Begin charging session | EV charger | | `STOP_CHARGING` | End charging session | EV charger | | `CHANGE_AVAILABILITY` | Enable/disable asset | EV charger | | `SET_SOC_TARGET` | Set target SOC | Battery | ## Event Publishing The dispatch engine publishes events at each lifecycle transition: | Action | Event Type | | ------------------ | -------------------- | | Commands generated | `SCHEDULE_RECEIVED` | | Command sent | `COMMAND_DISPATCHED` | | Command succeeded | `COMMAND_COMPLETED` | | Command failed | `COMMAND_FAILED` | | Command timed out | `COMMAND_TIMED_OUT` | ```python theme={null} # Monitor all dispatched commands bus.subscribe(EventType.COMMAND_DISPATCHED, lambda e: log.info( f"Dispatched {e.payload['command_type']} to {e.payload['asset_id']}" )) ``` ## Lookup Methods ```python theme={null} # Find a specific command cmd = engine.get_command("cmd_a1b2c3d4e5f6") # Get all commands for an asset asset_cmds = engine.get_commands_for_asset("ast_charger_001") ``` # Getting Started Source: https://docs.qubit.energy/layer-4/getting-started Install and dispatch your first optimization schedule in 5 minutes # Getting Started with Layer 4 Coordination Set up command dispatch and asset coordination in minutes. This guide walks you through converting a Layer 3 optimization schedule into real device commands. ## Installation ```bash theme={null} pip install qubit-energy-coordinator ``` ```python theme={null} from coordinator import ( DispatchEngine, EventBus, OCPPAdapter, ModbusAdapter, AdapterRegistry, create_ev_charger_state_machine, create_battery_state_machine ) print("Qubit Energy Coordinator installed successfully!") ``` ## Quick Examples ### Dispatch an EV Charging Schedule End-to-end: optimize with Layer 3, then dispatch with Layer 4. ```python Full Pipeline theme={null} from optimizer.ev.scheduler import EVChargingScheduler, ChargingSession from optimizer.base import OptimizationConfig from coordinator import ( DispatchEngine, EventBus, EventType, OCPPAdapter, AdapterRegistry, create_ev_charger_state_machine ) from datetime import datetime, timezone # --- Layer 3: Optimize --- config = OptimizationConfig(horizon="24h", resolution="1h") scheduler = EVChargingScheduler(config=config, site_capacity_kw=200.0, charger_capacity_kw=22.0) sessions = [ ChargingSession( vehicle_id="ev_001", arrival_time=datetime(2025, 1, 15, 8, 0, tzinfo=timezone.utc), departure_time=datetime(2025, 1, 15, 17, 0, tzinfo=timezone.utc), energy_needed_kwh=30.0, max_charge_rate_kw=22.0, ), ] tariff = { "energy_rates": [ {"name": "off_peak", "rate": 0.08, "schedule": {"start_time": "00:00", "end_time": "06:59"}}, {"name": "peak", "rate": 0.25, "schedule": {"start_time": "07:00", "end_time": "22:59"}}, {"name": "off_peak_night", "rate": 0.08, "schedule": {"start_time": "23:00", "end_time": "23:59"}} ] } result = scheduler.optimize( sessions=sessions, tariff=tariff, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc) ) print(f"Optimization: {result.status}, cost=${result.total_cost:.2f}") # --- Layer 4: Coordinate --- bus = EventBus() registry = AdapterRegistry() registry.register("ast_ev_001", OCPPAdapter("ocpp_site")) engine = DispatchEngine(event_bus=bus, adapter_registry=registry) # Convert schedule to commands commands = engine.schedule_to_commands( result.schedule, asset_map={"ev_001": "ast_ev_001"} ) print(f"Generated {len(commands)} commands") # Dispatch pending commands now = datetime(2025, 1, 15, 9, 0, tzinfo=timezone.utc) for cmd in engine.get_pending_commands(now): adapter = registry.get(cmd.asset_id) engine.dispatch_command(cmd, adapter) print(f" Dispatched: {cmd.command_type.value} → {cmd.asset_id} " f"({cmd.parameters['power_kw']:.0f} kW)") ``` ```python State Machine Integration theme={null} # Track charger state alongside dispatch charger = create_ev_charger_state_machine("ast_ev_001", event_bus=bus) # Simulate vehicle arrival charger.trigger("plug_in") print(f"Charger state: {charger.current_state}") # PREPARING charger.trigger("start_charge") print(f"Charger state: {charger.current_state}") # CHARGING # Check valid actions print(f"Available triggers: {charger.available_triggers()}") # ["suspend", "complete", "unplug", "fault"] # Vehicle departs charger.trigger("complete") charger.trigger("unplug") print(f"Charger state: {charger.current_state}") # AVAILABLE ``` ```python Event Monitoring theme={null} # Monitor all events def log_event(event): print(f"[{event.event_type.value}] {event.payload}") bus.subscribe(EventType.COMMAND_DISPATCHED, log_event) bus.subscribe(EventType.COMMAND_COMPLETED, log_event) bus.subscribe(EventType.STATE_CHANGED, log_event) # Events fire automatically during dispatch and state transitions ``` ### Battery Dispatch from Peak Shaving ```python Battery Dispatch theme={null} from optimizer.peak_shaving.controller import PeakShavingController, BatterySpec from optimizer.base import OptimizationConfig from coordinator import ( DispatchEngine, EventBus, ModbusAdapter, AdapterRegistry, create_battery_state_machine ) from datetime import datetime, timezone import numpy as np # --- Layer 3: Optimize --- config = OptimizationConfig(horizon="24h", resolution="1h") controller = PeakShavingController(config=config, peak_target_kw=150.0) battery_spec = BatterySpec( asset_id="ast_batt_001", capacity_kwh=200.0, max_charge_kw=50.0, max_discharge_kw=50.0, ) hours = np.arange(24) load = 80 + 100 * np.exp(-((hours - 9) ** 2) / 4) + 120 * np.exp(-((hours - 18) ** 2) / 4) result = controller.optimize( battery=battery_spec, load_forecast=load, start_time=datetime(2025, 1, 15, 0, 0, tzinfo=timezone.utc) ) print(f"Peak before: {load.max():.0f} kW → after: {result.peak_demand_kw:.0f} kW") # --- Layer 4: Coordinate --- bus = EventBus() registry = AdapterRegistry() registry.register("ast_batt_001", ModbusAdapter("modbus_bms")) engine = DispatchEngine(event_bus=bus, adapter_registry=registry) commands = engine.schedule_to_commands( result.schedule, asset_map={"battery": "ast_batt_001"} ) print(f"\nGenerated {len(commands)} battery commands:") for cmd in commands: print(f" {cmd.scheduled_at.strftime('%H:%M')} | " f"{cmd.command_type.value} | {cmd.parameters['power_kw']:.1f} kW") ``` ## Configuration Reference ### CoordinatorConfig | Parameter | Default | Description | | --------------------------- | -------- | ------------------------------- | | `site_id` | required | Qubit site identifier (`sit_*`) | | `dispatch_interval_seconds` | `5.0` | Seconds between dispatch cycles | | `command_timeout_seconds` | `30.0` | Default command timeout | | `max_retries` | `3` | Max retries for failed commands | | `retry_backoff_factor` | `2.0` | Exponential backoff multiplier | | `stale_state_seconds` | `60.0` | Asset state staleness threshold | ### Command Defaults | Parameter | Default | Description | | ----------------- | ------- | ----------------------------- | | `max_retries` | `3` | Retries before marking FAILED | | `timeout_seconds` | `30.0` | Seconds before TIMED\_OUT | | `priority` | `1` | 1=normal, higher=more urgent | ## Next Steps Deep dive into command lifecycle and change detection EV charger and battery state machine details OCPP, Modbus, and writing custom adapters Generate the schedules that Layer 4 dispatches *** *You're now ready to dispatch optimization schedules to real energy assets! Layer 4 coordinates the execution of Layer 3's intelligence.* # Layer 4 Overview Source: https://docs.qubit.energy/layer-4/overview Command dispatch, state machines, and protocol translation for energy asset coordination # Layer 4: Coordination Engine [![Production Ready](https://img.shields.io/badge/status-production-green)](https://github.com/qubit-foundation/qubit-energy-coordinator) [![Python](https://img.shields.io/badge/python-3.9%2B-blue)](https://www.python.org/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) Layer 4 bridges the gap between Layer 3's optimization schedules and real-world energy assets. It converts time-indexed dispatch plans into device-specific commands, manages asset state lifecycles, and provides event-driven coordination across the system. ## Available Features Converts Layer 3 OptimizationResult schedules into timed commands with lifecycle tracking and retry logic OCPP-aligned EV charger and battery state machines with transition guards and event publishing OCPP adapter for EV chargers, Modbus adapter for batteries, with an extensible registry pattern Typed synchronous pub/sub system for real-time coordination events across all components ## Quick Start ### Installation ```bash theme={null} pip install qubit-energy-coordinator ``` ### Dispatch an Optimization Schedule ```python theme={null} from coordinator import DispatchEngine, EventBus, OCPPAdapter, AdapterRegistry # Set up infrastructure bus = EventBus() registry = AdapterRegistry() registry.register("ast_ev_001", OCPPAdapter("ocpp_site_1")) engine = DispatchEngine(event_bus=bus, adapter_registry=registry) # Convert a Layer 3 schedule to commands commands = engine.schedule_to_commands( optimization_result.schedule, asset_map={"ev_001": "ast_ev_001"} ) # Dispatch pending commands for cmd in engine.get_pending_commands(now): adapter = registry.get(cmd.asset_id) engine.dispatch_command(cmd, adapter) ``` ## Architecture ```mermaid theme={null} graph TD A[Layer 3 Schedule] --> B[Dispatch Engine] B --> C[Command Queue] C --> D{Protocol Adapter} D -->|OCPP| E[EV Chargers] D -->|Modbus| F[Battery Systems] G[State Machines] --> B B --> H[Event Bus] H --> I[Subscribers] subgraph "Coordination Layer" B C G H end ``` ## Core Components ### Dispatch Engine The central orchestration component. It reads an `OptimizationResult.schedule` DataFrame (from Layer 3) and produces timed `Command` objects: * **EV columns** (`ev_{vehicle_id}_kw`) become `SET_CHARGE_RATE` commands * **Battery columns** (`battery_charge_kw`, `battery_discharge_kw`) become `SET_CHARGE_RATE` / `SET_DISCHARGE_RATE` commands * Only emits commands on value changes (avoids redundant setpoints) * Tracks command lifecycle: `PENDING → DISPATCHED → COMPLETED` (or `FAILED` / `TIMED_OUT`) * Automatic retry with configurable backoff ### State Machines Generic finite state machine with energy asset presets: OCPP-aligned lifecycle with 7 states: ``` AVAILABLE → PREPARING → CHARGING → FINISHING → AVAILABLE ↕ ↕ SUSPENDED FAULTED ↕ UNAVAILABLE ``` Triggers: `plug_in`, `start_charge`, `suspend`, `resume`, `complete`, `unplug`, `fault`, `reset`, `disable`, `enable` 5-state battery management: ``` IDLE ↔ CHARGING ↔ DISCHARGING ↕ STANDBY FAULTED ``` Supports direct mode switching (charge → discharge without stopping). Triggers: `charge`, `discharge`, `stop`, `standby`, `activate`, `fault`, `reset` ### Protocol Adapters Translate abstract commands into device-specific payloads: Maps command types to OCPP 1.6/2.0 actions: | Command Type | OCPP Action | | --------------------- | ------------------------ | | `SET_CHARGE_RATE` | `SetChargingProfile` | | `START_CHARGING` | `RemoteStartTransaction` | | `STOP_CHARGING` | `RemoteStopTransaction` | | `CHANGE_AVAILABILITY` | `ChangeAvailability` | Generates complete OCPP JSON payloads including charging profiles with power limits in watts. Maps commands to Modbus register writes: | Command Type | Default Register | | -------------------- | ---------------- | | `SET_CHARGE_RATE` | 40001 | | `SET_DISCHARGE_RATE` | 40002 | | `SET_POWER_LIMIT` | 40003 | | `SET_SOC_TARGET` | 40004 | Custom register maps supported per deployment. ### Event Bus Typed synchronous pub/sub with 8 event types: ```python theme={null} from coordinator import EventBus, EventType bus = EventBus() # Subscribe to state changes bus.subscribe(EventType.STATE_CHANGED, lambda e: print( f"Asset {e.payload['asset_id']}: {e.payload['from_state']} → {e.payload['to_state']}" )) # Subscribe to command failures for alerting bus.subscribe(EventType.COMMAND_FAILED, lambda e: alert_oncall(e.payload)) ``` Event types: `COMMAND_CREATED`, `COMMAND_DISPATCHED`, `COMMAND_ACKNOWLEDGED`, `COMMAND_COMPLETED`, `COMMAND_FAILED`, `COMMAND_TIMED_OUT`, `STATE_CHANGED`, `SCHEDULE_RECEIVED` ## Integration with Qubit Stack Layer 3 produces an `OptimizationResult` with a schedule DataFrame containing per-asset power setpoints `DispatchEngine.schedule_to_commands()` parses the DataFrame columns and creates timed `Command` objects State machines are checked before dispatch — commands to faulted or unavailable assets are held The `AdapterRegistry` finds the correct protocol adapter and translates the command to device-specific format Commands are sent to assets, responses tracked, and events published for monitoring and re-optimization triggers ## Next Steps Install and dispatch your first schedule in 5 minutes Deep dive into schedule-to-command conversion and lifecycle management EV charger and battery state machine details OCPP, Modbus, and custom adapter development Explore source code and contribute *** *Layer 4 Coordination Engine bridges optimization intelligence with physical energy assets, enabling real-time dispatch and control.* # Protocol Adapters Source: https://docs.qubit.energy/layer-4/protocol-adapters OCPP, Modbus, and custom protocol translation for energy asset control # Protocol Adapters Protocol adapters translate abstract Qubit commands into device-specific message formats. Each adapter handles the encoding, transport semantics, and response parsing for a specific protocol. ## Adapter Pattern All adapters extend the `ProtocolAdapter` base class: ```python theme={null} from coordinator import ProtocolAdapter, Command, AssetState class MyCustomAdapter(ProtocolAdapter): def send_command(self, command: Command) -> bool: """Translate and send. Return True on success.""" payload = self._translate(command) return self._send(payload) def read_state(self, asset_id: str) -> AssetState: """Read current device state.""" raw = self._read_registers(asset_id) return self._parse_state(raw) ``` ## OCPP Adapter For EV charging stations running OCPP 1.6 or 2.0: ```python theme={null} from coordinator import OCPPAdapter adapter = OCPPAdapter("ocpp_site_1") ``` ### Command Mapping | Qubit Command | OCPP Action | Payload | | --------------------- | ------------------------ | ------------------------------------------- | | `SET_CHARGE_RATE` | `SetChargingProfile` | Charging schedule with power limit in watts | | `SET_POWER_LIMIT` | `SetChargingProfile` | Same, used for dynamic power management | | `START_CHARGING` | `RemoteStartTransaction` | Connector ID + ID tag | | `STOP_CHARGING` | `RemoteStopTransaction` | Transaction ID | | `CHANGE_AVAILABILITY` | `ChangeAvailability` | Operative / Inoperative | ### Example: Set Charging Profile ```python theme={null} from coordinator.base import Command, CommandType cmd = Command( asset_id="ast_charger_001", command_type=CommandType.SET_CHARGE_RATE, parameters={"power_kw": 15.0, "connector_id": 1}, scheduled_at=now ) adapter.send_command(cmd) # After send_command, the OCPP payload is attached: print(cmd.parameters["ocpp_action"]) # "SetChargingProfile" print(cmd.parameters["ocpp_payload"]) # { # "connectorId": 1, # "csChargingProfiles": { # "chargingSchedule": { # "chargingRateUnit": "W", # "chargingSchedulePeriod": [{"startPeriod": 0, "limit": 15000.0}] # } # } # } ``` ## Modbus Adapter For battery management systems and inverters using Modbus TCP/RTU: ```python theme={null} from coordinator import ModbusAdapter # Default register map adapter = ModbusAdapter("modbus_bms_1") # Custom register map for specific BMS adapter = ModbusAdapter("modbus_bms_1", register_map={ "charge_rate_kw": 30001, "discharge_rate_kw": 30002, "power_limit_kw": 30003, "soc_target_percent": 30004, }) ``` ### Register Mapping | Qubit Command | Default Register | Value | | -------------------- | ---------------- | ----------- | | `SET_CHARGE_RATE` | 40001 | Power in kW | | `SET_DISCHARGE_RATE` | 40002 | Power in kW | | `SET_POWER_LIMIT` | 40003 | Power in kW | | `SET_SOC_TARGET` | 40004 | Percent | ### Example: Battery Discharge ```python theme={null} cmd = Command( asset_id="ast_batt_001", command_type=CommandType.SET_DISCHARGE_RATE, parameters={"power_kw": 50.0}, scheduled_at=now ) adapter.send_command(cmd) print(cmd.parameters["modbus_writes"]) # [(40002, 50.0)] ``` ## Adapter Registry The `AdapterRegistry` maps asset IDs to their protocol adapters: ```python theme={null} from coordinator import AdapterRegistry, OCPPAdapter, ModbusAdapter registry = AdapterRegistry() # Register adapters registry.register("ast_charger_001", OCPPAdapter("ocpp_1")) registry.register("ast_charger_002", OCPPAdapter("ocpp_1")) registry.register("ast_batt_001", ModbusAdapter("modbus_1")) # Look up by asset ID adapter = registry.get("ast_charger_001") # Check registration "ast_charger_001" in registry # True len(registry) # 3 # Remove registry.unregister("ast_charger_002") ``` If an unregistered asset ID is requested, a `KeyError` is raised with diagnostic info: ```python theme={null} registry.get("ast_unknown") # KeyError: "No protocol adapter registered for asset 'ast_unknown'. # Registered assets: ['ast_charger_001', 'ast_batt_001']" ``` ## Writing Custom Adapters To add support for a new protocol (e.g., DNP3, IEC 61850, REST API): ```python theme={null} from coordinator import ProtocolAdapter, Command, AssetState, AssetType class DNP3Adapter(ProtocolAdapter): def __init__(self, adapter_id: str, outstation_address: int): super().__init__(adapter_id) self.outstation_address = outstation_address def send_command(self, command: Command) -> bool: # Translate to DNP3 direct operate dnp3_payload = { "function": "DIRECT_OPERATE", "address": self.outstation_address, "index": self._command_to_index(command.command_type), "value": command.parameters.get("power_kw", 0), } command.parameters["dnp3_payload"] = dnp3_payload return True def read_state(self, asset_id: str) -> AssetState: return AssetState( asset_id=asset_id, asset_type=AssetType.INVERTER, current_state="RUNNING", ) # Register it registry.register("ast_inverter_001", DNP3Adapter("dnp3_1", outstation_address=10)) ``` # State Machines Source: https://docs.qubit.energy/layer-4/state-machines Asset lifecycle management with OCPP-aligned EV charger and battery state machines # State Machines The coordination layer uses finite state machines to track and guard asset lifecycles. Each controllable asset has a state machine instance that ensures commands are only dispatched when the asset is in a valid state. ## Generic State Machine The `StateMachine` class is a reusable, configurable FSM: ```python theme={null} from coordinator import StateMachine, EventBus sm = StateMachine( asset_id="ast_custom_001", states={"OFF", "STARTING", "RUNNING", "STOPPING"}, initial_state="OFF", transitions={ "OFF": {"start": "STARTING"}, "STARTING": {"ready": "RUNNING", "fail": "OFF"}, "RUNNING": {"stop": "STOPPING"}, "STOPPING": {"stopped": "OFF"}, }, event_bus=EventBus() # Optional ) sm.trigger("start") # → "STARTING" sm.can_trigger("ready") # → True sm.available_triggers() # → ["ready", "fail"] ``` ### Transition Guards Invalid transitions raise `InvalidTransition` with diagnostic info: ```python theme={null} from coordinator import InvalidTransition try: sm.trigger("stop") # Not valid from STARTING except InvalidTransition as e: print(e.current_state) # "STARTING" print(e.trigger) # "stop" print(e.valid_triggers) # ["ready", "fail"] ``` ### History Tracking Every transition is recorded: ```python theme={null} sm.trigger("start") sm.trigger("ready") print(sm.history) # [{"from_state": "OFF", "to_state": "STARTING", "trigger": "start", "timestamp": ...}, # {"from_state": "STARTING", "to_state": "RUNNING", "trigger": "ready", "timestamp": ...}] ``` ## EV Charger State Machine OCPP-aligned lifecycle with 7 states: ```mermaid theme={null} stateDiagram-v2 AVAILABLE --> PREPARING : plug_in AVAILABLE --> UNAVAILABLE : disable AVAILABLE --> FAULTED : fault PREPARING --> CHARGING : start_charge PREPARING --> AVAILABLE : unplug PREPARING --> FAULTED : fault CHARGING --> SUSPENDED : suspend CHARGING --> FINISHING : complete CHARGING --> AVAILABLE : unplug CHARGING --> FAULTED : fault SUSPENDED --> CHARGING : resume SUSPENDED --> FINISHING : complete SUSPENDED --> AVAILABLE : unplug SUSPENDED --> FAULTED : fault FINISHING --> AVAILABLE : unplug FINISHING --> FAULTED : fault FAULTED --> AVAILABLE : reset UNAVAILABLE --> AVAILABLE : enable UNAVAILABLE --> FAULTED : fault ``` ```python theme={null} from coordinator import create_ev_charger_state_machine charger = create_ev_charger_state_machine("ast_charger_001") charger.trigger("plug_in") # AVAILABLE → PREPARING charger.trigger("start_charge") # PREPARING → CHARGING charger.trigger("complete") # CHARGING → FINISHING charger.trigger("unplug") # FINISHING → AVAILABLE ``` ### Key Scenarios `AVAILABLE → PREPARING → CHARGING → FINISHING → AVAILABLE` The happy path: vehicle plugs in, charging starts, completes, vehicle unplugs. `CHARGING → SUSPENDED → CHARGING` Charging paused (e.g., grid signal, user request) then resumed. `CHARGING → AVAILABLE` (via `unplug`) Vehicle unplugs before charging completes. The dispatch engine should detect this and mark remaining commands as cancelled. `Any State → FAULTED → AVAILABLE` (via `reset`) Hardware fault detected. After reset, charger returns to available. ## Battery State Machine 5-state battery management with direct mode switching: ```mermaid theme={null} stateDiagram-v2 IDLE --> CHARGING : charge IDLE --> DISCHARGING : discharge IDLE --> STANDBY : standby IDLE --> FAULTED : fault CHARGING --> IDLE : stop CHARGING --> DISCHARGING : discharge CHARGING --> FAULTED : fault DISCHARGING --> IDLE : stop DISCHARGING --> CHARGING : charge DISCHARGING --> FAULTED : fault STANDBY --> IDLE : activate STANDBY --> FAULTED : fault FAULTED --> IDLE : reset ``` ```python theme={null} from coordinator import create_battery_state_machine battery = create_battery_state_machine("ast_batt_001") battery.trigger("charge") # IDLE → CHARGING battery.trigger("discharge") # CHARGING → DISCHARGING (direct switch) battery.trigger("stop") # DISCHARGING → IDLE ``` ### Direct Mode Switching The battery state machine supports direct transitions between charging and discharging without requiring a stop in between. This enables the dispatch engine to switch modes immediately when the optimization schedule transitions from charge to discharge. ## Event Bus Integration All state machines can publish `STATE_CHANGED` events: ```python theme={null} from coordinator import EventBus, EventType, create_battery_state_machine bus = EventBus() bus.subscribe(EventType.STATE_CHANGED, lambda e: print( f"{e.payload['asset_id']}: {e.payload['from_state']} → {e.payload['to_state']}" )) battery = create_battery_state_machine("ast_batt_001", event_bus=bus) battery.trigger("charge") # Prints: ast_batt_001: IDLE → CHARGING ``` # Layer 5 Overview Source: https://docs.qubit.energy/layer-5/overview Cryptographically secure settlement and verification for energy transactions # Layer 5: Settlement Layer *In Research Phase - Publishing 2026* Layer 5 provides cryptographically secure systems for energy transaction verification, automated settlement, and transparent economic coordination across the energy ecosystem. ## Planned Features Blockchain-based renewable energy and carbon credit tracking Smart contract-based payment and revenue distribution Transparent carbon footprint verification and offsetting Automated regulatory reporting and audit trails ## Research Areas Current research includes: * **Zero-knowledge proof systems** for privacy-preserving settlements * **Multi-chain interoperability** for cross-platform transactions * **Cryptographic verification** of energy generation and consumption * **Decentralized identity** for energy prosumers and assets ## Integration Points Layer 5 will provide settlement for: * **Layer 4**: Coordinated energy transactions and services * **Layer 6**: User-facing payment and billing interfaces * **External systems**: Traditional utility billing and regulatory compliance *** *The settlement layer ensures trust and transparency in the new energy economy. Follow our research progress on [GitHub](https://github.com/qubit-foundation).* # Layer 6 Overview Source: https://docs.qubit.energy/layer-6/overview User experience and API layer for energy applications # Layer 6: Interface Layer *In Planning Phase - Publishing 2026* Layer 6 provides intuitive interfaces and APIs that make energy data and capabilities accessible to end users, developers, and third-party applications. ## Planned Features RESTful and GraphQL APIs with comprehensive SDKs Mobile and web applications for energy prosumers Real-time grid and asset management interfaces Third-party integration endpoints and webhooks ## Planned Components The interface layer will include: * **Unified API Gateway** with authentication and rate limiting * **Real-time WebSocket streams** for live data * **Mobile SDKs** for iOS and Android applications * **Web Components** for easy embedding in existing systems * **Analytics Dashboards** with customizable visualizations ## Integration Points Layer 6 will provide interfaces for: * **All layers**: Unified access to data, predictions, optimization, and settlement * **External applications**: Third-party energy management systems * **End users**: Consumers, operators, and developers *** *The interface layer will make the full power of the Qubit Foundation stack accessible to everyone. Stay tuned for updates as we begin development.*