1-- Grid frequency monitoring with 1-second windows
 2WITH GridFrequency AS (
 3    SELECT 
 4        SubstationId,
 5        AVG(Frequency) as AvgFrequency,
 6        MIN(Frequency) as MinFrequency,
 7        MAX(Frequency) as MaxFrequency,
 8        System.Timestamp() as WindowEnd
 9    FROM TelemetryInput TIMESTAMP BY EventTime
10    GROUP BY SubstationId, TumblingWindow(second, 1)
11)
12SELECT * INTO FrequencyAlerts
13FROM GridFrequency
14WHERE AvgFrequency < 49.8 OR AvgFrequency > 50.2

 1public class GridOptimizationService : BackgroundService
 2{
 3    private readonly IHubContext<GridMonitoringHub> _hubContext;
 4    private readonly IDeviceControlService _deviceControl;
 5    private readonly IGridAnalyzer _analyzer;
 6
 7    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 8    {
 9        await foreach (var measurement in _telemetryStream.ReadAllAsync(stoppingToken))
10        {
11            var analysis = await _analyzer.AnalyzeFrequencyStability(measurement);
12            
13            if (analysis.RequiresImmediateAction)
14            {
15                // Analyze trends and dispatch high-level (non-real-time) control actions
16                var controlAction = GenerateControlAction(analysis);
17                await _deviceControl.ExecuteAsync(controlAction);
18                
19                // Notify operators in real-time
20                await _hubContext.Clients.Group($"grid-{measurement.Region}")
21                  .SendAsync("GridAnomalyDetected", new
22                    {
23                        Severity = analysis.Severity,
24                        Action = controlAction.Description,
25                        Timestamp = DateTime.UtcNow,
26                        PredictedImpact = analysis.PredictedImpact
27                    }, stoppingToken);
28            }
29        }
30    }
31
32    private ControlAction GenerateControlAction(GridAnalysis analysis)
33    {
34        return analysis.AnomalyType switch
35        {
36            AnomalyType.OverFrequency => new ControlAction
37            {
38                Type = ControlActionType.LoadIncrease,
39                Magnitude = CalculateLoadAdjustment(analysis.FrequencyDeviation),
40                TargetDevices = SelectOptimalLoads(analysis.AffectedRegion)
41            },
42            AnomalyType.UnderFrequency => new ControlAction
43            {
44                Type = ControlActionType.GenerationIncrease,
45                Magnitude = CalculateGenerationAdjustment(analysis.FrequencyDeviation),
46                TargetDevices = SelectAvailableGenerators(analysis.AffectedRegion)
47            },
48            _ => ControlAction.NoAction
49        };
50    }
51}

 1[ApiController]
 2
 3public class DeviceController : ControllerBase
 4{
 5    private readonly IDeviceService _deviceService;
 6    private readonly DaprClient _daprClient; // Injected Dapr Client
 7    private readonly ILogger<DeviceController> _logger;
 8
 9    public DeviceController(IDeviceService deviceService, DaprClient daprClient, ILogger<DeviceController> logger)
10    {
11        _deviceService = deviceService;
12        _daprClient = daprClient;
13        _logger = logger;
14    }
15
16    [HttpPost("{deviceId}/telemetry")]
17    public async Task<IActionResult> ReceiveTelemetry(
18        string deviceId, TelemetryData data)
19    {
20        var stopwatch = Stopwatch.StartNew();
21        
22        try
23        {
24            // Validate device and data integrity
25            var validationResult = await _deviceService.ValidateTelemetryAsync(deviceId, data);
26            if (!validationResult.IsValid)
27                return BadRequest(validationResult.Errors);
28
29            // Process telemetry with enrichment
30            var enrichedData = await _deviceService.EnrichTelemetryAsync(data);
31            
32            // Publish using Dapr for downstream processing
33            await _daprClient.PublishEventAsync("pubsub", "telemetry-received", new TelemetryReceived
34            {
35                DeviceId = deviceId,
36                Data = enrichedData,
37                Timestamp = DateTime.UtcNow,
38                CorrelationId = HttpContext.TraceIdentifier
39            });
40
41            stopwatch.Stop();
42            
43            return Ok(new 
44            { 
45                Status = "Processed", 
46                ProcessingLatency = stopwatch.ElapsedMilliseconds,
47                DataPoints = enrichedData.Measurements.Count
48            });
49        }
50        catch (Exception ex)
51        {
52            _logger.LogError(ex, "Error processing telemetry for device {DeviceId}", deviceId);
53            return StatusCode(500, "Processing error");
54        }
55    }
56
57    [HttpGet("{deviceId}/health")]
58    public async Task<IActionResult> GetDeviceHealth(string deviceId)
59    {
60        var health = await _deviceService.GetHealthStatusAsync(deviceId);
61        return Ok(health);
62    }
63}

 1public class OptimizedTelemetryRepository
 2{
 3    // Inject IServiceProvider or NpgsqlDataSource (Npgsql 7+) to resolve scoped
 4    // connections, not a singleton NpgsqlConnection.
 5    private readonly IServiceProvider _serviceProvider;
 6
 7    public OptimizedTelemetryRepository(IServiceProvider serviceProvider)
 8    {
 9        _serviceProvider = serviceProvider;
10    }
11
12    public async Task<IEnumerable<TelemetryReading>> GetAggregatedReadingsAsync(
13        string deviceId, DateTime from, DateTime to, TimeSpan interval)
14    {
15        // Create/dispose connection for each call. The pool makes this fast and thread-safe.
16        await using var scope = _serviceProvider.CreateAsyncScope();
17        await using var connection = scope.ServiceProvider.GetRequiredService<NpgsqlConnection>();
18        await connection.OpenAsync();
19
20        // Use time bucket aggregation for efficient querying
21        const string sql = @"
22            SELECT 
23                time_bucket(@interval, timestamp) AS bucket,
24                device_id,
25                AVG(voltage) as avg_voltage,
26                AVG(current) as avg_current,
27                AVG(power_factor) as avg_power_factor,
28                MIN(voltage) as min_voltage,
29                MAX(voltage) as max_voltage,
30                COUNT(*) as sample_count
31            FROM telemetry_readings 
32            WHERE device_id = @deviceId 
33              AND timestamp >= @from 
34              AND timestamp <= @to
35            GROUP BY bucket, device_id
36            ORDER BY bucket DESC";
37
38        await using var command = new NpgsqlCommand(sql, connection);
39        command.Parameters.AddWithValue("deviceId", deviceId);
40        command.Parameters.AddWithValue("from", from);
41        command.Parameters.AddWithValue("to", to);
42        command.Parameters.AddWithValue("interval", interval);
43
44        var readings = new List<TelemetryReading>();
45        await using var reader = await command.ExecuteReaderAsync();
46        
47        while (await reader.ReadAsync())
48        {
49            readings.Add(new TelemetryReading
50            {
51                DeviceId = reader.GetString("device_id"),
52                Timestamp = reader.GetDateTime("bucket"),
53                AvgVoltage = reader.GetDouble("avg_voltage"),
54                //... map other fields
55            });
56        }
57        
58        return readings;
59    }
60
61    public async Task BulkInsertOptimizedAsync(IEnumerable<TelemetryReading> readings)
62    {
63        await using var scope = _serviceProvider.CreateAsyncScope();
64        await using var connection = scope.ServiceProvider.GetRequiredService<NpgsqlConnection>();
65        await connection.OpenAsync();
66        
67        // Use binary COPY for maximum insert performance
68        const string copyCommand = 
69            "COPY telemetry_readings (device_id, timestamp, voltage, current, power_factor) " +
70            "FROM STDIN (FORMAT BINARY)";
71
72        await using var writer = await connection.BeginBinaryImportAsync(copyCommand);
73
74        foreach (var reading in readings)
75        {
76            await writer.StartRowAsync();
77            await writer.WriteAsync(reading.DeviceId);
78            await writer.WriteAsync(reading.Timestamp, NpgsqlDbType.TimestampTz);
79            await writer.WriteAsync(reading.Voltage);
80            await writer.WriteAsync(reading.Current);
81            await writer.WriteAsync(reading.PowerFactor);
82        }
83
84        await writer.CompleteAsync();
85    }
86}

 1public class GridProtectionModule : ModuleClient
 2{
 3    private readonly ConcurrentDictionary<string, DeviceState> _deviceStates = new();
 4    private readonly Timer _protectionTimer;
 5    private readonly ILogger _logger;
 6
 7    public GridProtectionModule(ILogger logger)
 8    {
 9        _logger = logger;
10        // 5ms monitoring cycle for high-frequency polling
11        _protectionTimer = new Timer(ExecuteProtectionCycle, null, 
12            TimeSpan.Zero, TimeSpan.FromMilliseconds(5));
13    }
14
15    private async void ExecuteProtectionCycle(object state)
16    {
17        var cycleStart = DateTime.UtcNow;
18        
19        try
20        {
21            // Logic here *monitors* state from dedicated hardware (IEDs)
22            foreach (var (deviceId, deviceState) in _deviceStates)
23            {
24                await EvaluateProtectionLogic(deviceId, deviceState);
25            }
26        }
27        catch (Exception ex)
28        {
29            _logger.LogError(ex, "Protection cycle error");
30        }
31        
32        var cycleTime = DateTime.UtcNow - cycleStart;
33        if (cycleTime.TotalMilliseconds > 5) // Alert if cycle exceeds monitoring interval
34        {
35            _logger.LogWarning("Protection cycle time exceeded: {CycleTime}ms", 
36                cycleTime.TotalMilliseconds);
37        }
38    }
39
40    private async Task EvaluateProtectionLogic(string deviceId, DeviceState state)
41    {
42        // Example: Overcurrent protection logic
43        if (state.Current > state.InstantaneousThreshold)
44        {
45            await TripBreakerAsync(deviceId, "Instantaneous Overcurrent");
46            return;
47        }
48        //... other protection logic
49    }
50
51   
52    public async Task<MethodResponse> UpdateMeasurements(MethodRequest methodRequest)
53    {
54        var measurements = JsonSerializer.Deserialize<DeviceMeasurements>(methodRequest.DataAsJson);
55        
56        _deviceStates.AddOrUpdate(measurements.DeviceId, 
57            new DeviceState(measurements), 
58            (key, existing) => existing.UpdateMeasurements(measurements));
59        
60        return new MethodResponse(200);
61    }
62
63    private async Task TripBreakerAsync(string deviceId, string reason)
64    {
65        var tripCommand = new
66        {
67            DeviceId = deviceId,
68            Action = "TRIP",
69            Reason = reason,
70            Timestamp = DateTime.UtcNow.Ticks
71        };
72
73        // Forward trip command to dedicated relay hardware via a separate module
74        await SendEventAsync("protection-trips", 
75            new Message(JsonSerializer.SerializeToUtf8Bytes(tripCommand)));
76
77        _logger.LogCritical("Circuit breaker trip signal sent: {DeviceId}, Reason: {Reason}", 
78            deviceId, reason);
79    }
80}

 1public class EquipmentPredictionService
 2{
 3    // The _predictionEngine can be loaded from an ML.NET-trained model
 4    // or, more likely, an ONNX model exported from Python.
 5    private readonly PredictionEngine<EquipmentData, FailurePrediction> _predictionEngine;
 6    private readonly ILogger<EquipmentPredictionService> _logger;
 7
 8    public EquipmentPredictionService(PredictionEngine<EquipmentData, FailureFailurePrediction> predictionEngine, 
 9                                      ILogger<EquipmentPredictionService> logger)
10    {
11        _predictionEngine = predictionEngine;
12        _logger = logger;
13    }
14
15    public async Task<EquipmentHealthAssessment> AssessEquipmentHealthAsync(
16        string equipmentId, TelemetryData recentData)
17    {
18        // Feature engineering from telemetry data
19        var features = ExtractFeatures(recentData);
20        var equipmentData = new EquipmentData
21        {
22            EquipmentId = equipmentId,
23            Temperature = features.AvgTemperature,
24            Vibration = features.VibrationRms,
25            //... other features
26        };
27
28        // Generate prediction
29        var prediction = _predictionEngine.Predict(equipmentData);
30
31        // Calculate time to failure estimate
32        var timeToFailure = EstimateTimeToFailure(prediction.Probability, features);
33
34        return new EquipmentHealthAssessment
35        {
36            EquipmentId = equipmentId,
37            HealthScore = 1.0f - prediction.Probability,
38            FailureProbability = prediction.Probability,
39            EstimatedTimeToFailure = timeToFailure,
40            RiskLevel = DetermineRiskLevel(prediction.Probability),
41            RecommendedActions = GenerateRecommendations(prediction, timeToFailure),
42            ConfidenceLevel = prediction.Score,
43            AssessmentTimestamp = DateTime.UtcNow
44        };
45    }
46
47    private EquipmentFeatures ExtractFeatures(TelemetryData data)
48    {
49        //... feature extraction logic
50        return new EquipmentFeatures();
51    }
52
53    private List<MaintenanceRecommendation> GenerateRecommendations(
54        FailurePrediction prediction, TimeSpan timeToFailure)
55    {
56        //... recommendation logic
57        return new List<MaintenanceRecommendation>();
58    }
59}

 1public class PartitionedTelemetryProcessor
 2{
 3    private readonly IServiceProvider _serviceProvider;
 4    private readonly int _partitionCount;
 5
 6    public async Task RouteToProcessorAsync(string deviceId, TelemetryData data)
 7    {
 8        // Consistent hashing ensures even distribution
 9        var partitionKey = CalculatePartition(deviceId);
10        var processor = _serviceProvider.GetKeyedService<ITelemetryProcessor>(partitionKey);
11        
12        await processor.ProcessAsync(data);
13    }
14
15    private int CalculatePartition(string deviceId)
16    {
17        // Use device region prefix for locality
18        var regionHash = deviceId.Substring(0, 3).GetHashCode();
19        return Math.Abs(regionHash) % _partitionCount;
20    }
21}