SCADA Substation Event Ingestion and Cleansing Script

Role: Senior Data Engineer specializing in substation telemetry, grid automation, and SCADA data pipelines.

Context

• Grid operator: {{grid_operator}}
• Ingestion source format: {{source_stream_type}}
• Target warehouse database: {{target_warehouse}}
• Target staging schema: {{raw_event_schema}}
• Batch record threshold: {{batch_size_records}}
• Maximum allowable malformed rate: {{max_error_rate_pct}}%

Task

Create a resilient Python database ingestion script that extracts raw SCADA telemetry from {{source_stream_type}}, validates electrical sensor values, deduplicates message IDs, and batch-loads sanitized events into {{target_warehouse}} under schema {{raw_event_schema}}.

Method

1. Configure database connection pooling and environment-driven credentials for {{target_warehouse}}.
2. Parse incoming {{source_stream_type}} records into structured records (substation_id, feeder_id, voltage, frequency, timestamp).
3. Implement validation filters for electrical anomalies (e.g., negative frequency, out-of-band voltage surges).
4. Deduplicate incoming event payloads based on composite key (substation_id, event_uuid, timestamp).
5. Direct anomalous and malformed payloads into a dedicated quarantine dead-letter staging table.
6. Terminate pipeline execution if malformed records exceed {{max_error_rate_pct}}%.
7. Execute bulk database inserts using parameter arrays when memory reaches {{batch_size_records}} items.
8. Return execution statistics including processed count, quarantined count, and elapsed runtime.

Constraints

• MUST write PEP-8 compliant Python 3.10+ using standard libraries and SQLAlchemy/psycopg.
• MUST NOT hardcode database credentials or grid connection strings.
• All database writes MUST execute inside an explicit database transaction with rollback support.
• Include type hints for all function signatures.

Output format

Provide the script in the following format:

• Config & Imports Block
• Schema Definition DDL (DDL for main table & quarantine table)
• Main Ingestion & Sanitization Python Class
• Batch Write & Transaction Context Manager
• Entrypoint main() execution stub Keep the total code length under 180 lines.

Self-review

1. Ensure memory consumption remains constant regardless of stream length.
2. Confirm malformed payload rate calculation triggers failure when exceeding {{max_error_rate_pct}}%.
3. Validate that {{target_warehouse}} bulk write API uses parameter binding.