Innovenergy_trunk/csharp/App/Collector/src/Program.cs

using System.Net.Sockets;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text.Json;
using InnovEnergy.App.Collector.Influx;
using InnovEnergy.App.Collector.Records;
using InnovEnergy.Lib.Utils;
using InnovEnergy.Lib.Utils.Net;
using InnovEnergy.Lib.WebServer;
using static System.Text.Encoding;
using static InnovEnergy.Lib.Utils.ExceptionHandling;


namespace InnovEnergy.App.Collector;

// dotnet publish Collector.csproj -c Release -r linux-x64 -p:PublishTrimmed=false -p:PublishSingleFile=true --self-contained true ; scp ./bin/Release/net6.0/linux-x64/publish/* ig@salidomo.innovenergy.ch:~/collector

internal record BatteryData
(
    String Installation, 
    String Battery, 
    Double EnergyCapacity, 
    Double EnergyStored,
    Double Power
);


internal static class Program
{
    //private static readonly Logger Logger = new Logger(Settings.LoggingEndPoint);

    private static UdpClient _incomingSocket = new UdpClient(Settings.IncomingEndPoint);
    private static UdpClient _dbSocket       = new UdpClient();
    
    private static readonly Subject<BatteryData>  Batteries   = new Subject<BatteryData>();
    private static readonly JsonSerializerOptions JsonOptions = new JsonSerializerOptions { WriteIndented = true };
    
    public static void Main(String[] args)
    {
        Task.Run(ServeJsonStats);

        while (true)
            ProcessDatagram();

        // ReSharper disable once FunctionNeverReturns
    }

    private static void ServeJsonStats()
    {
        var json = "";
        
        Batteries.ObserveOn(TaskPoolScheduler.Default)
                 .Buffer(TimeSpan.FromSeconds(4))
                 .Select(b => b.GroupBy(d => d.Battery).Select(d => d.First()).ToList())
                 .Select(ToJson)
                 .Subscribe(j => json = j);

        HttpResponse ServeRequest(HttpRequest httpRequest)
        {
            return new HttpResponse
            {
                Content     = json.Apply(UTF8.GetBytes),
                ContentType = ContentType.ApplicationJson,
                Headers     = new[] { new HttpHeader("Access-Control-Allow-Origin", "*") }
            };
        }

        WebServer.ServeOnLocalHost(3333, ServeRequest);
    }

    private static String ToJson(IReadOnlyCollection<BatteryData> batteryData)
    {
        var nInstallations   = batteryData.GroupBy(d => d.Installation).Count();
        var nBatteries       = batteryData.Count;
        var energyStored     = batteryData.Sum(d => d.EnergyStored).Apply(Math.Round);
        var energyCapacity   = batteryData.Sum(d => d.EnergyCapacity).Apply(Math.Round);
        var chargingPower    = batteryData.Where(d => d.Power > 0).Sum(d => d.Power / 1000).Apply(Math.Round);
        var dischargingPower = batteryData.Where(d => d.Power < 0).Sum(d => -d.Power / 1000).Apply(Math.Round);

        var json = new
        {
            nInstallations,
            nBatteries,
            energyStored_kWh    = energyStored,
            energyCapacity_kWh  = energyCapacity,
            chargingPower_kW    = chargingPower,
            dischargingPower_kW = dischargingPower,
        };

        return JsonSerializer.Serialize(json, JsonOptions);
        
        
        // Console.WriteLine($"nInstallations  : {nInstallations}");
        // Console.WriteLine($"nBatteries      : {nBatteries}");
        // Console.WriteLine($"energyStored    : {Math.Round(energyStored)} kWh");
        // Console.WriteLine($"energyCapacity  : {Math.Round(energyCapacity)} kWh");
        // Console.WriteLine($"chargingPower   : {Math.Round(chargingPower / 1000)} kW");
        // Console.WriteLine($"dischargingPower: {Math.Round(dischargingPower/ 1000)} kW");
        
    }


    private static void ProcessDatagram()
    {
        ReadDatagram()
            .ThenTry(ParseDatagram)
            .ThenTry(SendToDb)
            .OnErrorDo(Console.WriteLine);
    }

    private static Try<Byte[]> ParseDatagram(UdpDatagram datagram)
    {
        Byte[] Parse()
        {
            var batteryRecords = BatteryDataParser
                                .ParseDatagram(datagram);

            if (batteryRecords.FirstOrDefault() is BatteryStatus bs)
            {
                var battery      = bs.InstallationName + bs.BatteryId;
                var capacity     = 48.0/1000 * bs.AmpereHours;
                var energyStored = (Double) bs.Soc / 100 * capacity;
                var power        = bs.Current * bs.Voltage;

                var data = new BatteryData(bs.InstallationName, 
                                           battery, 
                                           capacity,
                                           energyStored,
                                           (Double)power);
                
                Batteries.OnNext(data);
            }
                
            return batteryRecords
                  .Select(InfluxRecord.Serialize)
                  .JoinLines()
                  .Apply(UTF8.GetBytes);
        }

        return Try(Parse)
              .OnErrorLog("ParseDatagram failed " + datagram.EndPoint.Address);
    }

    private static Try<UdpDatagram> ReadDatagram()
    {
        return Try(_incomingSocket.ReadDatagram)
              .OnErrorLog("Failed to read from UDP socket")
              .OnErrorDo(ResetIncomingSocket);
    }


    private static Try<Int32> SendToDb(Byte[] data)
    {
        Int32 Send() => _dbSocket.SendDatagram(data, Settings.DbEndPoint);

        return Try(Send)
              .OnErrorLog("SendToDb failed")
              .OnErrorDo(ResetDbSocket);
    }

    private static void ResetDbSocket(Exception e)
    {
        _dbSocket.Dispose();
        _dbSocket = new UdpClient();
    }

    private static void ResetIncomingSocket(Exception e)
    {
        _incomingSocket.Dispose();
        _incomingSocket = new UdpClient(Settings.IncomingEndPoint);
    }
}