using System.Collections.Immutable; using System.Reactive.Linq; using System.Reactive.Threading.Tasks; using InnovEnergy.Lib.Protocols.DBus; using InnovEnergy.Lib.Protocols.DBus.Daemon; using InnovEnergy.Lib.Protocols.DBus.Protocol.DataTypes; using InnovEnergy.Lib.Utils; namespace InnovEnergy.Lib.Victron.VeDBus; using VeProps = ImmutableDictionary<String, VeProperty>; public record struct ServiceProperties(String ServiceName, VeProps Properties); public static class VeDBus { public static IObservable<VeProps> ObservePropertiesOfService(this DBusConnection connection, String serviceName) { return connection .ObservePropertiesOfServices(busName => busName == serviceName) .SelectMany(s => s.Take(1)) .Select(s => s.Properties); } public static IObservable<IReadOnlyList<ServiceProperties>> ObservePropertiesOfServices(this DBusConnection connection, Func<String, Boolean> serviceSelector) { // subscription management could probably be done using a combination of GroupBy and Switch. // however I dont understand GroupBy :( var services = new Dictionary<String, (IObservable<VeProps> props, IDisposable subscription)>(); return connection .Services .Where(c => serviceSelector(c.BusNameOrId)) .Where(c => c.Change != ServiceChange.NoChange) .Select(ObserveProps) .Switch() .OnDispose(DisposeSubscriptions); IObservable<IReadOnlyList<ServiceProperties>> ObserveProps(ServiceChanged serviceChanged) { var (change, serviceName, _, _) = serviceChanged; lock (services) { if (services.TryGetValue(serviceName, out var sv)) { sv.subscription.Dispose(); services.Remove(serviceName); } if (change is ServiceChange.BusNameAdded or ServiceChange.IdAdded) { var props = connection .ObserveAllProperties(serviceName) .Replay(1); services.Add(serviceName, (props, props.Connect())); } var observeProps = services.Select(kv => kv.Value.props).ToList(); // ToList is important! var serviceNames = services.Select(kv => kv.Key).ToList(); // ToList is important! return Observable .CombineLatest(observeProps) .Select(props => Enumerable .Zip(serviceNames, props, (s, ps) => new ServiceProperties(s, ps)) .ToList()); } } void DisposeSubscriptions() { lock (services) { foreach (var service in services.Values) service.subscription.Dispose(); services.Clear(); } } } public static IObservable<VeProperty> ObservePropertiesChanged(this DBusConnection connection, String service , ObjectPath? objectPath = default) { // some VeServices dont set the interface (!) so don't use it in the filter return connection.ObserveSignalMessages(sender: service, objectPath: objectPath, member: VeDBusApi.PropertiesChanged) .TrySelect(VeProperty.Decode); } private static IObservable<VeProps> ObserveAllProperties(this DBusConnection connection, String serviceName) { var init = connection .GetAllProperties(serviceName) .Catch(Observable.Empty<VeProperty>()) .Aggregate(VeProps.Empty, UpdateItem); // produces a single aggregated dict return init .Select(ObserveChanges) .Switch(); VeProps UpdateItem(VeProps props, VeProperty prop) => props.SetItem(prop.ObjectPath, prop); IObservable<VeProps> ObserveChanges(VeProps initialProps) { return connection .ObservePropertiesChanged(serviceName) .Scan(initialProps, UpdateItem) .StartWith(initialProps); // not sure if necessary } } public static IObservable<VeProperty> GetAllProperties(this DBusConnection connection, String serviceName) { var texts = connection.GetAllTexts(serviceName) .ToObservable(); var values = connection.GetAllValues(serviceName).ToObservable(); return Observable .Zip(values, texts, Merge) .SelectMany(ps => ps); static IEnumerable<VeProperty> Merge(IReadOnlyDictionary<String, Object> values, IReadOnlyDictionary<String, String> texts) { return from kv in values let path = kv.Key let value = kv.Value let text = texts.TryGetValue(path, out var txt) ? txt : value.ToString() select new VeProperty(path, value, text); } } public static async Task<IReadOnlyDictionary<String, Object>> GetAllValues(this DBusConnection connection, String destination) { var result = await connection.CallMethod<Object>(destination: destination, @interface: VeDBusApi.Interface, objectPath: ObjectPath.Root, member: VeDBusApi.GetValue); while (result is Variant v) result = v.Value; return result is IReadOnlyDictionary<String, Variant> variantDict ? variantDict.ToDictionary(kv => kv.Key, kv => kv.Value.Value) : new Dictionary<String, Object>(); } public static async Task<IReadOnlyDictionary<String, String>> GetAllTexts(this DBusConnection connection, String destination) { var result = await connection.CallMethod<Object>(destination: destination, @interface: VeDBusApi.Interface, objectPath: ObjectPath.Root, member: VeDBusApi.GetText); while (result is Variant v) result = v.Value; return result as IReadOnlyDictionary<String, String> ?? new Dictionary<String, String>(); } public static async Task<T> GetValue<T>(this DBusConnection connection, String destination, ObjectPath objectPath) { var value = await connection.GetValue(destination, objectPath); return (T)value; } public static async Task<Object> GetValue(this DBusConnection connection, String destination, ObjectPath objectPath) { var variant = await connection.CallMethod<Variant>(destination: destination, @interface: VeDBusApi.Interface, objectPath: objectPath, member: VeDBusApi.GetValue); return variant.Value; } public static Task<String> GetText(this DBusConnection connection, String destination, ObjectPath objectPath) { return connection.CallMethod<String>(destination: destination, @interface: VeDBusApi.Interface, objectPath: objectPath, member: VeDBusApi.GetText); } public static async Task<Boolean> SetValue<T>(this DBusConnection connection, String destination, ObjectPath objectPath, T value) { var setValue = await connection.CallMethod<Int32>(destination: destination, @interface: VeDBusApi.Interface, objectPath: objectPath, member: VeDBusApi.SetValue, payload: value!.Variant()); return setValue == 0; } public static void BroadcastPropertiesChanged(this DBusConnection connection, ObjectPath path, Object value, String text) { var veProp = new VeProperty(path, value, text); connection.BroadcastPropertiesChanged(veProp); } public static void BroadcastPropertiesChanged(this DBusConnection connection, VeProperty property) { var veProp = new Dictionary<String, Variant> { { "Value", property.Value.Variant() }, { "Text" , property.Text.Variant() } }; connection.BroadcastSignal(VeDBusApi.Interface, property.ObjectPath, VeDBusApi.PropertiesChanged, veProp); } }