forked from victronenergy/dbus-systemcalc-py
-
Notifications
You must be signed in to change notification settings - Fork 0
/
dbus_systemcalc.py
executable file
·1049 lines (920 loc) · 41.5 KB
/
dbus_systemcalc.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/python3 -u
# -*- coding: utf-8 -*-
from dbus.mainloop.glib import DBusGMainLoop
import dbus
import argparse
import sys
import os
import json
import time
import re
from gi.repository import GLib
# Victron packages
sys.path.insert(1, os.path.join(os.path.dirname(__file__), 'ext', 'velib_python'))
from vedbus import VeDbusService
from ve_utils import get_vrm_portal_id, exit_on_error
from dbusmonitor import DbusMonitor
from settingsdevice import SettingsDevice
from logger import setup_logging
import delegates
from sc_utils import safeadd as _safeadd, safemax as _safemax
softwareVersion = '2.92'
class SystemCalc:
STATE_IDLE = 0
STATE_CHARGING = 1
STATE_DISCHARGING = 2
BATSERVICE_DEFAULT = 'default'
BATSERVICE_NOBATTERY = 'nobattery'
def __init__(self):
# Why this dummy? Because DbusMonitor expects these values to be there, even though we don't
# need them. So just add some dummy data. This can go away when DbusMonitor is more generic.
dummy = {'code': None, 'whenToLog': 'configChange', 'accessLevel': None}
dbus_tree = {
'com.victronenergy.solarcharger': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/0/Current': dummy,
'/Load/I': dummy,
'/FirmwareVersion': dummy},
'com.victronenergy.pvinverter': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Ac/L1/Power': dummy,
'/Ac/L2/Power': dummy,
'/Ac/L3/Power': dummy,
'/Position': dummy,
'/ProductId': dummy},
'com.victronenergy.battery': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/DeviceInstance': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/1/Voltage': dummy,
'/Dc/0/Current': dummy,
'/Dc/0/Power': dummy,
'/Soc': dummy,
'/Sense/Current': dummy,
'/TimeToGo': dummy,
'/ConsumedAmphours': dummy,
'/ProductId': dummy,
'/CustomName': dummy},
'com.victronenergy.vebus' : {
'/Ac/ActiveIn/ActiveInput': dummy,
'/Ac/ActiveIn/L1/P': dummy,
'/Ac/ActiveIn/L2/P': dummy,
'/Ac/ActiveIn/L3/P': dummy,
'/Ac/Out/L1/P': dummy,
'/Ac/Out/L2/P': dummy,
'/Ac/Out/L3/P': dummy,
'/Connected': dummy,
'/ProductId': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Mode': dummy,
'/State': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/0/Current': dummy,
'/Dc/0/Power': dummy,
'/Soc': dummy},
'com.victronenergy.fuelcell': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/0/Current': dummy},
'com.victronenergy.charger': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/0/Current': dummy,
'/Dc/1/Voltage': dummy,
'/Dc/1/Current': dummy,
'/Dc/2/Voltage': dummy,
'/Dc/2/Current': dummy},
'com.victronenergy.grid' : {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/ProductId' : dummy,
'/DeviceType' : dummy,
'/Ac/L1/Power': dummy,
'/Ac/L2/Power': dummy,
'/Ac/L3/Power': dummy},
'com.victronenergy.genset' : {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/ProductId' : dummy,
'/DeviceType' : dummy,
'/Ac/L1/Power': dummy,
'/Ac/L2/Power': dummy,
'/Ac/L3/Power': dummy,
'/StarterVoltage': dummy},
'com.victronenergy.settings' : {
'/Settings/SystemSetup/AcInput1' : dummy,
'/Settings/SystemSetup/AcInput2' : dummy,
'/Settings/CGwacs/RunWithoutGridMeter' : dummy,
'/Settings/System/TimeZone' : dummy},
'com.victronenergy.temperature': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy},
'com.victronenergy.inverter': {
'/Connected': dummy,
'/ProductName': dummy,
'/Mgmt/Connection': dummy,
'/Dc/0/Voltage': dummy,
'/Dc/0/Current': dummy,
'/Ac/Out/L1/P': dummy,
'/Ac/Out/L1/V': dummy,
'/Ac/Out/L1/I': dummy,
'/Yield/Power': dummy,
'/Soc': dummy},
'com.victronenergy.dcsystem': {
'/Dc/0/Voltage': dummy,
'/Dc/0/Power': dummy
}
}
self._modules = [
delegates.Multi(),
delegates.HubTypeSelect(),
delegates.VebusSocWriter(),
delegates.ServiceMapper(),
delegates.RelayState(),
delegates.BuzzerControl(),
delegates.LgCircuitBreakerDetect(),
delegates.Dvcc(self),
delegates.BatterySense(self),
delegates.BatterySettings(self),
delegates.SystemState(self),
delegates.BatteryLife(),
delegates.ScheduledCharging(),
delegates.SourceTimers(),
#delegates.BydCurrentSense(self),
delegates.BatteryData(),
delegates.Gps(),
delegates.AcInputs(),
delegates.GensetStartStop(),
delegates.SocSync(self)]
for m in self._modules:
for service, paths in m.get_input():
s = dbus_tree.setdefault(service, {})
for path in paths:
s[path] = dummy
self._dbusmonitor = self._create_dbus_monitor(dbus_tree, valueChangedCallback=self._dbus_value_changed,
deviceAddedCallback=self._device_added, deviceRemovedCallback=self._device_removed)
# Connect to localsettings
supported_settings = {
'batteryservice': ['/Settings/SystemSetup/BatteryService', self.BATSERVICE_DEFAULT, 0, 0],
'hasdcsystem': ['/Settings/SystemSetup/HasDcSystem', 0, 0, 1],
'useacout': ['/Settings/SystemSetup/HasAcOutSystem', 1, 0, 1]}
for m in self._modules:
for setting in m.get_settings():
supported_settings[setting[0]] = list(setting[1:])
self._settings = self._create_settings(supported_settings, self._handlechangedsetting)
self._dbusservice = self._create_dbus_service()
for m in self._modules:
m.set_sources(self._dbusmonitor, self._settings, self._dbusservice)
# At this moment, VRM portal ID is the MAC address of the CCGX. Anyhow, it should be string uniquely
# identifying the CCGX.
self._dbusservice.add_path('/Serial', value=get_vrm_portal_id())
self._dbusservice.add_path(
'/AvailableBatteryServices', value=None, gettextcallback=self._gettext)
self._dbusservice.add_path(
'/AvailableBatteryMeasurements', value=None)
self._dbusservice.add_path(
'/AutoSelectedBatteryService', value=None, gettextcallback=self._gettext)
self._dbusservice.add_path(
'/AutoSelectedBatteryMeasurement', value=None, gettextcallback=self._gettext)
self._dbusservice.add_path(
'/ActiveBatteryService', value=None, gettextcallback=self._gettext)
self._dbusservice.add_path(
'/Dc/Battery/BatteryService', value=None)
self._dbusservice.add_path(
'/PvInvertersProductIds', value=None)
self._summeditems = {
'/Ac/Grid/L1/Power': {'gettext': '%.0F W'},
'/Ac/Grid/L2/Power': {'gettext': '%.0F W'},
'/Ac/Grid/L3/Power': {'gettext': '%.0F W'},
'/Ac/Grid/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/Grid/ProductId': {'gettext': '%s'},
'/Ac/Grid/DeviceType': {'gettext': '%s'},
'/Ac/Genset/L1/Power': {'gettext': '%.0F W'},
'/Ac/Genset/L2/Power': {'gettext': '%.0F W'},
'/Ac/Genset/L3/Power': {'gettext': '%.0F W'},
'/Ac/Genset/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/Genset/ProductId': {'gettext': '%s'},
'/Ac/Genset/DeviceType': {'gettext': '%s'},
'/Ac/ConsumptionOnOutput/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnOutput/L1/Power': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnOutput/L2/Power': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnOutput/L3/Power': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnInput/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnInput/L1/Power': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnInput/L2/Power': {'gettext': '%.0F W'},
'/Ac/ConsumptionOnInput/L3/Power': {'gettext': '%.0F W'},
'/Ac/Consumption/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/Consumption/L1/Power': {'gettext': '%.0F W'},
'/Ac/Consumption/L2/Power': {'gettext': '%.0F W'},
'/Ac/Consumption/L3/Power': {'gettext': '%.0F W'},
'/Ac/Consumption/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/PvOnOutput/L1/Power': {'gettext': '%.0F W'},
'/Ac/PvOnOutput/L2/Power': {'gettext': '%.0F W'},
'/Ac/PvOnOutput/L3/Power': {'gettext': '%.0F W'},
'/Ac/PvOnOutput/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/PvOnGrid/L1/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGrid/L2/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGrid/L3/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGrid/NumberOfPhases': {'gettext': '%.0F W'},
'/Ac/PvOnGenset/L1/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGenset/L2/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGenset/L3/Power': {'gettext': '%.0F W'},
'/Ac/PvOnGenset/NumberOfPhases': {'gettext': '%d'},
'/Dc/Pv/Power': {'gettext': '%.0F W'},
'/Dc/Pv/Current': {'gettext': '%.1F A'},
'/Dc/Battery/Voltage': {'gettext': '%.2F V'},
'/Dc/Battery/VoltageService': {'gettext': '%s'},
'/Dc/Battery/Current': {'gettext': '%.1F A'},
'/Dc/Battery/Power': {'gettext': '%.0F W'},
'/Dc/Battery/Soc': {'gettext': '%.0F %%'},
'/Dc/Battery/State': {'gettext': '%s'},
'/Dc/Battery/TimeToGo': {'gettext': '%.0F s'},
'/Dc/Battery/ConsumedAmphours': {'gettext': '%.1F Ah'},
'/Dc/Battery/ProductId': {'gettext': '0x%x'},
'/Dc/Charger/Power': {'gettext': '%.0F %%'},
'/Dc/FuelCell/Power': {'gettext': '%.0F %%'},
'/Dc/Vebus/Current': {'gettext': '%.1F A'},
'/Dc/Vebus/Power': {'gettext': '%.0F W'},
'/Dc/System/Power': {'gettext': '%.0F W'},
'/Dc/System/MeasurementType': {'gettext': '%d'},
'/Ac/ActiveIn/Source': {'gettext': '%s'},
'/Ac/ActiveIn/L1/Power': {'gettext': '%.0F W'},
'/Ac/ActiveIn/L2/Power': {'gettext': '%.0F W'},
'/Ac/ActiveIn/L3/Power': {'gettext': '%.0F W'},
'/Ac/ActiveIn/NumberOfPhases': {'gettext': '%d'},
}
for m in self._modules:
self._summeditems.update(m.get_output())
for path in self._summeditems.keys():
self._dbusservice.add_path(path, value=None, gettextcallback=self._gettext)
self._batteryservice = None
self._determinebatteryservice()
if self._batteryservice is None:
logger.info("Battery service initialized to None (setting == %s)" %
self._settings['batteryservice'])
self._changed = True
for service, instance in self._dbusmonitor.get_service_list().items():
self._device_added(service, instance, do_service_change=False)
self._handleservicechange()
self._updatevalues()
GLib.timeout_add(1000, exit_on_error, self._handletimertick)
def _create_dbus_monitor(self, *args, **kwargs):
raise Exception("This function should be overridden")
def _create_settings(self, *args, **kwargs):
raise Exception("This function should be overridden")
def _create_dbus_service(self):
raise Exception("This function should be overridden")
def _handlechangedsetting(self, setting, oldvalue, newvalue):
self._determinebatteryservice()
self._changed = True
# Give our delegates a chance to react on a settings change
for m in self._modules:
m.settings_changed(setting, oldvalue, newvalue)
def _find_device_instance(self, serviceclass, instance):
""" Gets a mapping of services vs DeviceInstance using
get_service_list. Then searches for the specified DeviceInstance
and returns the service name. """
services = self._dbusmonitor.get_service_list(classfilter=serviceclass)
for k, v in services.items():
if v == instance:
return k
return None
def _determinebatteryservice(self):
auto_battery_service = self._autoselect_battery_service()
auto_battery_measurement = None
auto_selected = False
if auto_battery_service is not None:
services = self._dbusmonitor.get_service_list()
if auto_battery_service in services:
auto_battery_measurement = \
self._get_instance_service_name(auto_battery_service, services[auto_battery_service])
auto_battery_measurement = auto_battery_measurement.replace('.', '_').replace('/', '_') + '/Dc/0'
self._dbusservice['/AutoSelectedBatteryMeasurement'] = auto_battery_measurement
if self._settings['batteryservice'] == self.BATSERVICE_DEFAULT:
auto_selected = True
newbatteryservice = auto_battery_service
self._dbusservice['/AutoSelectedBatteryService'] = (
'No battery monitor found' if newbatteryservice is None else
self._get_readable_service_name(newbatteryservice))
elif self._settings['batteryservice'] == self.BATSERVICE_NOBATTERY:
self._dbusservice['/AutoSelectedBatteryService'] = None
newbatteryservice = None
else:
self._dbusservice['/AutoSelectedBatteryService'] = None
s = self._settings['batteryservice'].split('/')
if len(s) != 2:
logger.error("The battery setting (%s) is invalid!" % self._settings['batteryservice'])
serviceclass = s[0]
instance = int(s[1]) if len(s) == 2 else None
# newbatteryservice might turn into None if a chosen battery
# monitor no longer exists. Don't auto change the setting (it might
# come back) and don't autoselect another.
newbatteryservice = self._find_device_instance(serviceclass, instance)
if newbatteryservice != self._batteryservice:
services = self._dbusmonitor.get_service_list()
instance = services.get(newbatteryservice, None)
if instance is None:
battery_service = None
else:
battery_service = self._get_instance_service_name(newbatteryservice, instance)
self._dbusservice['/ActiveBatteryService'] = battery_service
logger.info("Battery service, setting == %s, changed from %s to %s (%s)" %
(self._settings['batteryservice'], self._batteryservice, newbatteryservice, instance))
# Battery service has changed. Notify delegates.
for m in self._modules:
m.battery_service_changed(auto_selected, self._batteryservice, newbatteryservice)
self._dbusservice['/Dc/Battery/BatteryService'] = self._batteryservice = newbatteryservice
def _autoselect_battery_service(self):
# Default setting business logic:
# first try to use a battery service (BMV or Lynx Shunt VE.Can). If there
# is more than one battery service, just use a random one. If no battery service is
# available, check if there are not Solar chargers and no normal chargers. If they are not
# there, assume this is a hub-2, hub-3 or hub-4 system and use VE.Bus SOC.
batteries = self._get_connected_service_list('com.victronenergy.battery')
# Pick the first battery service
if len(batteries) > 0:
return sorted(batteries)[0]
# No battery services, and there is a charger in the system. Abandon
# hope.
if self._get_first_connected_service('com.victronenergy.charger') is not None:
return None
# Also no Multi, then give up.
vebus_service = self._get_service_having_lowest_instance('com.victronenergy.vebus')
if vebus_service is None:
# No VE.Bus, but maybe there is an inverter with built-in SOC
# tracking, eg RS Smart.
inverter = self._get_service_having_lowest_instance('com.victronenergy.inverter')
if inverter and self._dbusmonitor.get_value(inverter[0], '/Soc') is not None:
return inverter[0]
return None
# There is a Multi, it supports tracking external charge current from
# solarchargers, and there are no DC loads. Then use it.
if self._dbusmonitor.get_value(
vebus_service[0], '/ExtraBatteryCurrent') is not None \
and self._get_first_connected_service('com.victronenergy.dcsystem') is None \
and self._settings['hasdcsystem'] == 0:
return vebus_service[0]
# Multi does not support tracking solarcharger current, and we have
# solar chargers. Then we cannot use it.
if self._get_first_connected_service('com.victronenergy.solarcharger') is not None:
return None
# Only a Multi, no other chargers. Then we can use it.
return vebus_service[0]
@property
def batteryservice(self):
return self._batteryservice
# Called on a one second timer
def _handletimertick(self):
if self._changed:
self._updatevalues()
self._changed = False
return True # keep timer running
def _updatepvinverterspidlist(self):
# Create list of connected pv inverters id's
pvinverters = self._dbusmonitor.get_service_list('com.victronenergy.pvinverter')
productids = []
for pvinverter in pvinverters:
pid = self._dbusmonitor.get_value(pvinverter, '/ProductId')
if pid is not None and pid not in productids:
productids.append(pid)
self._dbusservice['/PvInvertersProductIds'] = productids
def _updatevalues(self):
# ==== PREPARATIONS ====
newvalues = {}
# Set the user timezone
if 'TZ' not in os.environ:
tz = self._dbusmonitor.get_value('com.victronenergy.settings', '/Settings/System/TimeZone')
if tz is not None:
os.environ['TZ'] = tz
time.tzset()
# Determine values used in logic below
vebusses = self._dbusmonitor.get_service_list('com.victronenergy.vebus')
vebuspower = 0
for vebus in vebusses:
v = self._dbusmonitor.get_value(vebus, '/Dc/0/Voltage')
i = self._dbusmonitor.get_value(vebus, '/Dc/0/Current')
if v is not None and i is not None:
vebuspower += v * i
# ==== PVINVERTERS ====
pvinverters = self._dbusmonitor.get_service_list('com.victronenergy.pvinverter')
pos = {0: '/Ac/PvOnGrid', 1: '/Ac/PvOnOutput', 2: '/Ac/PvOnGenset'}
for pvinverter in pvinverters:
# Position will be None if PV inverter service has just been removed (after retrieving the
# service list).
position = pos.get(self._dbusmonitor.get_value(pvinverter, '/Position'))
if position is not None:
for phase in range(1, 4):
power = self._dbusmonitor.get_value(pvinverter, '/Ac/L%s/Power' % phase)
if power is not None:
path = '%s/L%s/Power' % (position, phase)
newvalues[path] = _safeadd(newvalues.get(path), power)
for path in pos.values():
self._compute_number_of_phases(path, newvalues)
# ==== SOLARCHARGERS ====
solarchargers = self._dbusmonitor.get_service_list('com.victronenergy.solarcharger')
solarcharger_batteryvoltage = None
solarcharger_batteryvoltage_service = None
solarchargers_charge_power = 0
solarchargers_loadoutput_power = None
for solarcharger in solarchargers:
v = self._dbusmonitor.get_value(solarcharger, '/Dc/0/Voltage')
if v is None:
continue
i = self._dbusmonitor.get_value(solarcharger, '/Dc/0/Current')
if i is None:
continue
l = self._dbusmonitor.get_value(solarcharger, '/Load/I', 0)
if l is not None:
if solarchargers_loadoutput_power is None:
solarchargers_loadoutput_power = l * v
else:
solarchargers_loadoutput_power += l * v
solarchargers_charge_power += v * i
# Note that this path is not in the _summeditems{}, making for it to not be
# published on D-Bus. Which fine. The only one needing it is the vebussocwriter-
# delegate.
if '/Dc/Pv/ChargeCurrent' not in newvalues:
newvalues['/Dc/Pv/ChargeCurrent'] = i
else:
newvalues['/Dc/Pv/ChargeCurrent'] += i
if '/Dc/Pv/Power' not in newvalues:
newvalues['/Dc/Pv/Power'] = v * _safeadd(i, l)
newvalues['/Dc/Pv/Current'] = _safeadd(i, l)
solarcharger_batteryvoltage = v
solarcharger_batteryvoltage_service = solarcharger
else:
newvalues['/Dc/Pv/Power'] += v * _safeadd(i, l)
newvalues['/Dc/Pv/Current'] += _safeadd(i, l)
# ==== FUELCELLS ====
fuelcells = self._dbusmonitor.get_service_list('com.victronenergy.fuelcell')
fuelcell_batteryvoltage = None
fuelcell_batteryvoltage_service = None
for fuelcell in fuelcells:
# Assume the battery connected to output 0 is the main battery
v = self._dbusmonitor.get_value(fuelcell, '/Dc/0/Voltage')
if v is None:
continue
fuelcell_batteryvoltage = v
fuelcell_batteryvoltage_service = fuelcell
i = self._dbusmonitor.get_value(fuelcell, '/Dc/0/Current')
if i is None:
continue
if '/Dc/FuelCell/Power' not in newvalues:
newvalues['/Dc/FuelCell/Power'] = v * i
else:
newvalues['/Dc/FuelCell/Power'] += v * i
# ==== CHARGERS ====
chargers = self._dbusmonitor.get_service_list('com.victronenergy.charger')
charger_batteryvoltage = None
charger_batteryvoltage_service = None
for charger in chargers:
# Assume the battery connected to output 0 is the main battery
v = self._dbusmonitor.get_value(charger, '/Dc/0/Voltage')
if v is None:
continue
charger_batteryvoltage = v
charger_batteryvoltage_service = charger
i = self._dbusmonitor.get_value(charger, '/Dc/0/Current')
if i is None:
continue
if '/Dc/Charger/Power' not in newvalues:
newvalues['/Dc/Charger/Power'] = v * i
else:
newvalues['/Dc/Charger/Power'] += v * i
# ==== VE.Direct Inverters ====
_vedirect_inverters = sorted((di, s) for s, di in self._dbusmonitor.get_service_list('com.victronenergy.inverter').items())
vedirect_inverters = [x[1] for x in _vedirect_inverters]
vedirect_inverter = None
if vedirect_inverters:
vedirect_inverter = vedirect_inverters[0]
# For RS Smart inverters, add PV to the yield
for i in vedirect_inverters:
pv_yield = self._dbusmonitor.get_value(i, "/Yield/Power")
if pv_yield is not None:
newvalues['/Dc/Pv/Power'] = newvalues.get('/Dc/Pv/Power', 0) + pv_yield
# Used lower down, possibly needed for battery values as well
dcsystems = self._dbusmonitor.get_service_list('com.victronenergy.dcsystem')
# ==== BATTERY ====
if self._batteryservice is not None:
batteryservicetype = self._batteryservice.split('.')[2]
assert batteryservicetype in ('battery', 'vebus', 'inverter')
newvalues['/Dc/Battery/Soc'] = self._dbusmonitor.get_value(self._batteryservice,'/Soc')
newvalues['/Dc/Battery/TimeToGo'] = self._dbusmonitor.get_value(self._batteryservice,'/TimeToGo')
newvalues['/Dc/Battery/ConsumedAmphours'] = self._dbusmonitor.get_value(self._batteryservice,'/ConsumedAmphours')
newvalues['/Dc/Battery/ProductId'] = self._dbusmonitor.get_value(self._batteryservice, '/ProductId')
if batteryservicetype in ('battery', 'inverter'):
newvalues['/Dc/Battery/Voltage'] = self._dbusmonitor.get_value(self._batteryservice, '/Dc/0/Voltage')
newvalues['/Dc/Battery/VoltageService'] = self._batteryservice
newvalues['/Dc/Battery/Current'] = self._dbusmonitor.get_value(self._batteryservice, '/Dc/0/Current')
newvalues['/Dc/Battery/Power'] = self._dbusmonitor.get_value(self._batteryservice, '/Dc/0/Power')
elif batteryservicetype == 'vebus':
vebus_voltage = self._dbusmonitor.get_value(self._batteryservice, '/Dc/0/Voltage')
vebus_current = self._dbusmonitor.get_value(self._batteryservice, '/Dc/0/Current')
vebus_power = None if vebus_voltage is None or vebus_current is None else vebus_current * vebus_voltage
newvalues['/Dc/Battery/Voltage'] = vebus_voltage
newvalues['/Dc/Battery/VoltageService'] = self._batteryservice
if self._settings['hasdcsystem'] == 1 or dcsystems:
# hasdcsystem will normally disqualify the multi from being
# auto-selected as battery monitor, so the only way we're
# here is if the user explicitly selected the multi as the
# battery service
newvalues['/Dc/Battery/Current'] = vebus_current
if vebus_power is not None:
newvalues['/Dc/Battery/Power'] = vebus_power
else:
battery_power = _safeadd(solarchargers_charge_power, vebus_power)
newvalues['/Dc/Battery/Current'] = battery_power / vebus_voltage if vebus_voltage is not None and vebus_voltage > 0 else None
newvalues['/Dc/Battery/Power'] = battery_power
p = newvalues.get('/Dc/Battery/Power', None)
if p is not None:
if p > 30:
newvalues['/Dc/Battery/State'] = self.STATE_CHARGING
elif p < -30:
newvalues['/Dc/Battery/State'] = self.STATE_DISCHARGING
else:
newvalues['/Dc/Battery/State'] = self.STATE_IDLE
else:
# The battery service is not a BMS/BMV or a suitable vebus. A
# suitable vebus is defined as one explicitly selected by the user,
# or one that was automatically selected for SOC tracking. We may
# however still have a VE.Bus, just not one that can accurately
# track SOC. If we have one, use it as voltage source. Otherwise
# try a solar charger, a charger, a vedirect inverter or a dcsource
# as fallbacks.
batteryservicetype = None
vebusses = self._dbusmonitor.get_service_list('com.victronenergy.vebus')
for vebus in vebusses:
v = self._dbusmonitor.get_value(vebus, '/Dc/0/Voltage')
s = self._dbusmonitor.get_value(vebus, '/State')
if v is not None and s not in (0, None):
newvalues['/Dc/Battery/Voltage'] = v
newvalues['/Dc/Battery/VoltageService'] = vebus
break # Skip the else below
else:
# No suitable vebus voltage, try other devices
if solarcharger_batteryvoltage is not None:
newvalues['/Dc/Battery/Voltage'] = solarcharger_batteryvoltage
newvalues['/Dc/Battery/VoltageService'] = solarcharger_batteryvoltage_service
elif charger_batteryvoltage is not None:
newvalues['/Dc/Battery/Voltage'] = charger_batteryvoltage
newvalues['/Dc/Battery/VoltageService'] = charger_batteryvoltage_service
elif fuelcell_batteryvoltage is not None:
newvalues['/Dc/Battery/Voltage'] = fuelcell_batteryvoltage
newvalues['/Dc/Battery/VoltageService'] = fuelcell_batteryvoltage_service
elif vedirect_inverter is not None:
v = self._dbusmonitor.get_value(vedirect_inverter, '/Dc/0/Voltage')
if v is not None:
newvalues['/Dc/Battery/Voltage'] = v
newvalues['/Dc/Battery/VoltageService'] = vedirect_inverter
elif dcsystems:
# Get voltage from first dcsystem
s = next(iter(dcsystems.keys()))
v = self._dbusmonitor.get_value(s, '/Dc/0/Voltage')
if v is not None:
newvalues['/Dc/Battery/Voltage'] = v
newvalues['/Dc/Battery/VoltageService'] = s
# We have no suitable battery monitor, so power and current data
# is not available. We can however calculate it from other values,
# if we have at least a battery voltage.
if '/Dc/Battery/Voltage' in newvalues:
dcsystempower = _safeadd(0, *(self._dbusmonitor.get_value(s,
'/Dc/0/Power', 0) for s in dcsystems))
if dcsystems or self._settings['hasdcsystem'] == 0:
# Either DC loads are monitored, or there are no
# unmonitored DC loads or chargers: derive battery watts
# and amps from vebus, solarchargers, chargers and measured
# loads.
p = solarchargers_charge_power + newvalues.get('/Dc/Charger/Power', 0) + vebuspower - dcsystempower
voltage = newvalues['/Dc/Battery/Voltage']
newvalues['/Dc/Battery/Current'] = p / voltage if voltage > 0 else None
newvalues['/Dc/Battery/Power'] = p
# ==== SYSTEM POWER ====
# Look for dcsytem devices, add them together. Otherwise, if enabled,
# calculate it
if dcsystems:
newvalues['/Dc/System/MeasurementType'] = 1 # measured
newvalues['/Dc/System/Power'] = 0
for meter in dcsystems:
newvalues['/Dc/System/Power'] = _safeadd(newvalues['/Dc/System/Power'],
self._dbusmonitor.get_value(meter, '/Dc/0/Power'))
elif self._settings['hasdcsystem'] == 1 and batteryservicetype == 'battery':
# Calculate power being generated/consumed by not measured devices in the network.
# For MPPTs, take all the power, including power going out of the load output.
# /Dc/System: positive: consuming power
# VE.Bus: Positive: current flowing from the Multi to the dc system or battery
# Solarcharger & other chargers: positive: charging
# battery: Positive: charging battery.
# battery = solarcharger + charger + ve.bus - system
battery_power = newvalues.get('/Dc/Battery/Power')
if battery_power is not None:
dc_pv_power = newvalues.get('/Dc/Pv/Power', 0)
charger_power = newvalues.get('/Dc/Charger/Power', 0)
fuelcell_power = newvalues.get('/Dc/FuelCell/Power', 0)
# If there are VE.Direct inverters, remove their power from the
# DC estimate. This is done using the AC value when the DC
# power values are not available.
inverter_power = 0
for i in vedirect_inverters:
inverter_current = self._dbusmonitor.get_value(i, '/Dc/0/Current')
if inverter_current is not None:
inverter_power += self._dbusmonitor.get_value(
i, '/Dc/0/Voltage', 0) * inverter_current
else:
inverter_power += self._dbusmonitor.get_value(
i, '/Ac/Out/L1/V', 0) * self._dbusmonitor.get_value(
i, '/Ac/Out/L1/I', 0)
newvalues['/Dc/System/MeasurementType'] = 0 # estimated
newvalues['/Dc/System/Power'] = dc_pv_power + charger_power + fuelcell_power + vebuspower - inverter_power - battery_power
elif self._settings['hasdcsystem'] == 1 and solarchargers_loadoutput_power is not None:
newvalues['/Dc/System/MeasurementType'] = 0 # estimated
newvalues['/Dc/System/Power'] = solarchargers_loadoutput_power
# ==== Vebus ====
multi_path = getattr(delegates.Multi.instance.multi, 'service', None)
if multi_path is not None:
dc_current = self._dbusmonitor.get_value(multi_path, '/Dc/0/Current')
newvalues['/Dc/Vebus/Current'] = dc_current
dc_power = self._dbusmonitor.get_value(multi_path, '/Dc/0/Power')
# Just in case /Dc/0/Power is not available
if dc_power == None and dc_current is not None:
dc_voltage = self._dbusmonitor.get_value(multi_path, '/Dc/0/Voltage')
if dc_voltage is not None:
dc_power = dc_voltage * dc_current
# Note that there is also vebuspower, which is the total DC power summed over all multis.
# However, this value cannot be combined with /Dc/Multi/Current, because it does not make sense
# to add the Dc currents of all multis if they do not share the same DC voltage.
newvalues['/Dc/Vebus/Power'] = dc_power
# ===== AC IN SOURCE =====
ac_in_source = None
if multi_path is None:
# Check if we have an non-VE.Bus inverter. If yes, then ActiveInput
# is disconnected.
if vedirect_inverter is not None:
ac_in_source = 240
else:
active_input = self._dbusmonitor.get_value(multi_path, '/Ac/ActiveIn/ActiveInput')
if active_input == 0xF0:
# Not connected
ac_in_source = 240
elif active_input is not None:
settings_path = '/Settings/SystemSetup/AcInput%s' % (active_input + 1)
ac_in_source = self._dbusmonitor.get_value('com.victronenergy.settings', settings_path)
newvalues['/Ac/ActiveIn/Source'] = ac_in_source
# ===== GRID METERS & CONSUMPTION ====
grid_meter = delegates.AcInputs.instance.gridmeter
genset_meter = delegates.AcInputs.instance.gensetmeter
# Make an educated guess as to what is being consumed from an AC source. If ac_in_source
# indicates grid, genset or shore, we use that. If the Multi is off, or disconnected through
# a relay assistant or otherwise, then assume the presence of a .grid or .genset service indicates
# presence of that AC source. If both are available, then give up. This decision making is here
# so the GUI has something to present even if the Multi is off.
ac_in_guess = ac_in_source
if ac_in_guess in (None, 0xF0):
if genset_meter is None and grid_meter is not None:
ac_in_guess = 1
elif grid_meter is None and genset_meter is not None:
ac_in_guess = 2
consumption = { "L1" : None, "L2" : None, "L3" : None }
for device_type, em, _types in (('Grid', grid_meter, (1, 3)), ('Genset', genset_meter, (2,))):
# If a grid meter is present we use values from it. If not, we look at the multi. If it has
# AcIn1 or AcIn2 connected to the grid, we use those values.
# com.victronenergy.grid.??? indicates presence of an energy meter used as grid meter.
# com.victronenergy.vebus.???/Ac/ActiveIn/ActiveInput: decides which whether we look at AcIn1
# or AcIn2 as possible grid connection.
uses_active_input = ac_in_source in _types
for phase in consumption:
p = None
pvpower = newvalues.get('/Ac/PvOn%s/%s/Power' % (device_type, phase))
if em is not None:
p = self._dbusmonitor.get_value(em.service, '/Ac/%s/Power' % phase)
# Compute consumption between energy meter and multi (meter power - multi AC in) and
# add an optional PV inverter on input to the mix.
c = None
if uses_active_input:
ac_in = self._dbusmonitor.get_value(multi_path, '/Ac/ActiveIn/%s/P' % phase)
if ac_in is not None:
c = _safeadd(c, -ac_in)
# If there's any power coming from a PV inverter in the inactive AC in (which is unlikely),
# it will still be used, because there may also be a load in the same ACIn consuming
# power, or the power could be fed back to the net.
c = _safeadd(c, p, pvpower)
consumption[phase] = _safeadd(consumption[phase], _safemax(0, c))
else:
if uses_active_input:
p = self._dbusmonitor.get_value(multi_path, '/Ac/ActiveIn/%s/P' % phase)
if p is not None:
consumption[phase] = _safeadd(0, consumption[phase])
# No relevant energy meter present. Assume there is no load between the grid and the multi.
# There may be a PV inverter present though (Hub-3 setup).
if pvpower != None:
p = _safeadd(p, -pvpower)
newvalues['/Ac/%s/%s/Power' % (device_type, phase)] = p
if ac_in_guess in _types:
newvalues['/Ac/ActiveIn/%s/Power' % (phase,)] = p
self._compute_number_of_phases('/Ac/%s' % device_type, newvalues)
self._compute_number_of_phases('/Ac/ActiveIn', newvalues)
product_id = None
device_type_id = None
if em is not None:
product_id = em.product_id
device_type_id = em.device_type
if product_id is None and uses_active_input:
product_id = self._dbusmonitor.get_value(multi_path, '/ProductId')
newvalues['/Ac/%s/ProductId' % device_type] = product_id
newvalues['/Ac/%s/DeviceType' % device_type] = device_type_id
# If we have an ESS system and RunWithoutGridMeter is set, there cannot be load on the AC-In, so it
# must be on AC-Out. Hence we do calculate AC-Out consumption even if 'useacout' is disabled.
# Similarly all load are by definition on the output if this is not an ESS system.
use_ac_out = \
self._settings['useacout'] == 1 or \
(multi_path is not None and self._dbusmonitor.get_value(multi_path, '/Hub4/AssistantId') not in (4, 5)) or \
self._dbusmonitor.get_value('com.victronenergy.settings', '/Settings/CGwacs/RunWithoutGridMeter') == 1
for phase in consumption:
c = None
if use_ac_out:
c = newvalues.get('/Ac/PvOnOutput/%s/Power' % phase)
if multi_path is None:
for inv in vedirect_inverters:
ac_out = self._dbusmonitor.get_value(inv, '/Ac/Out/%s/P' % phase)
# Some models don't show power, calculate it
if ac_out is None:
i = self._dbusmonitor.get_value(inv, '/Ac/Out/%s/I' % phase)
u = self._dbusmonitor.get_value(inv, '/Ac/Out/%s/V' % phase)
if None not in (i, u):
ac_out = i * u
c = _safeadd(c, ac_out)
else:
ac_out = self._dbusmonitor.get_value(multi_path, '/Ac/Out/%s/P' % phase)
c = _safeadd(c, ac_out)
c = _safemax(0, c)
newvalues['/Ac/ConsumptionOnOutput/%s/Power' % phase] = c
newvalues['/Ac/ConsumptionOnInput/%s/Power' % phase] = consumption[phase]
newvalues['/Ac/Consumption/%s/Power' % phase] = _safeadd(consumption[phase], c)
self._compute_number_of_phases('/Ac/Consumption', newvalues)
self._compute_number_of_phases('/Ac/ConsumptionOnOutput', newvalues)
self._compute_number_of_phases('/Ac/ConsumptionOnInput', newvalues)
for m in self._modules:
m.update_values(newvalues)
# ==== UPDATE DBUS ITEMS ====
with self._dbusservice as sss:
for path in self._summeditems.keys():
# Why the None? Because we want to invalidate things we don't have anymore.
sss[path] = newvalues.get(path, None)
def _handleservicechange(self):
# Update the available battery monitor services, used to populate the dropdown in the settings.
# Below code makes a dictionary. The key is [dbuserviceclass]/[deviceinstance]. For example
# "battery/245". The value is the name to show to the user in the dropdown. The full dbus-
# servicename, ie 'com.victronenergy.vebus.ttyO1' is not used, since the last part of that is not
# fixed. dbus-serviceclass name and the device instance are already fixed, so best to use those.
services = self._get_connected_service_list('com.victronenergy.vebus')
services.update(self._get_connected_service_list('com.victronenergy.battery'))
services.update({k: v for k, v in self._get_connected_service_list(
'com.victronenergy.inverter').items() if self._dbusmonitor.get_value(k, '/Soc') is not None})
ul = {self.BATSERVICE_DEFAULT: 'Automatic', self.BATSERVICE_NOBATTERY: 'No battery monitor'}
for servicename, instance in services.items():
key = self._get_instance_service_name(servicename, instance)
ul[key] = self._get_readable_service_name(servicename)
self._dbusservice['/AvailableBatteryServices'] = json.dumps(ul)
ul = {self.BATSERVICE_DEFAULT: 'Automatic', self.BATSERVICE_NOBATTERY: 'No battery monitor'}
# For later: for device supporting multiple Dc measurement we should add entries for /Dc/1 etc as
# well.
for servicename, instance in services.items():
key = self._get_instance_service_name(servicename, instance).replace('.', '_').replace('/', '_') + '/Dc/0'
ul[key] = self._get_readable_service_name(servicename)
self._dbusservice['/AvailableBatteryMeasurements'] = ul
self._determinebatteryservice()
self._updatepvinverterspidlist()
self._changed = True
def _get_readable_service_name(self, servicename):
return '%s on %s' % (
self._dbusmonitor.get_value(servicename, '/ProductName'),
self._dbusmonitor.get_value(servicename, '/Mgmt/Connection'))
def _get_instance_service_name(self, service, instance):
return '%s/%s' % ('.'.join(service.split('.')[0:3]), instance)
def _remove_unconnected_services(self, services):
# Workaround: because com.victronenergy.vebus is available even when there is no vebus product
# connected, remove any service that is not connected. Previously we used
# /State since mandatory path /Connected is not implemented in mk2dbus,
# but this has since been resolved.
for servicename in list(services.keys()):
if (self._dbusmonitor.get_value(servicename, '/Connected') != 1
or self._dbusmonitor.get_value(servicename, '/ProductName') is None
or self._dbusmonitor.get_value(servicename, '/Mgmt/Connection') is None):
del services[servicename]
def _dbus_value_changed(self, dbusServiceName, dbusPath, dict, changes, deviceInstance):
self._changed = True
# Workaround because com.victronenergy.vebus is available even when there is no vebus product
# connected.
if (dbusPath in ['/Connected', '/ProductName', '/Mgmt/Connection'] or
(dbusPath == '/State' and dbusServiceName.split('.')[0:3] == ['com', 'victronenergy', 'vebus'])):
self._handleservicechange()
# Track the timezone changes
if dbusPath == '/Settings/System/TimeZone':
tz = changes.get('Value')
if tz is not None:
os.environ['TZ'] = tz
def _device_added(self, service, instance, do_service_change=True):
if do_service_change:
self._handleservicechange()
for m in self._modules:
m.device_added(service, instance, do_service_change)
def _device_removed(self, service, instance):
self._handleservicechange()
for m in self._modules:
m.device_removed(service, instance)
def _gettext(self, path, value):
if path == '/Dc/Battery/State':
state = {self.STATE_IDLE: 'Idle', self.STATE_CHARGING: 'Charging',
self.STATE_DISCHARGING: 'Discharging'}
return state[value]
item = self._summeditems.get(path)
if item is not None:
return item['gettext'] % value
return str(value)
def _compute_number_of_phases(self, path, newvalues):
number_of_phases = None
for phase in range(1, 4):
p = newvalues.get('%s/L%s/Power' % (path, phase))
if p is not None:
number_of_phases = phase
newvalues[path + '/NumberOfPhases'] = number_of_phases
def _get_connected_service_list(self, classfilter=None):
services = self._dbusmonitor.get_service_list(classfilter=classfilter)
self._remove_unconnected_services(services)
return services
# returns a servicename string
def _get_first_connected_service(self, classfilter):
services = self._get_connected_service_list(classfilter=classfilter)
if len(services) == 0:
return None
return next(iter(services.items()), (None,))[0]
# returns a tuple (servicename, instance)
def _get_service_having_lowest_instance(self, classfilter=None):
services = self._get_connected_service_list(classfilter=classfilter)
if len(services) == 0:
return None
# sort the dict by value; returns list of tuples: (value, key)
s = sorted((value, key) for (key, value) in services.items())
return (s[0][1], s[0][0])
class DbusSystemCalc(SystemCalc):
def _create_dbus_monitor(self, *args, **kwargs):
return DbusMonitor(*args, **kwargs)
def _create_settings(self, *args, **kwargs):
bus = dbus.SessionBus() if 'DBUS_SESSION_BUS_ADDRESS' in os.environ else dbus.SystemBus()
return SettingsDevice(bus, *args, timeout=10, **kwargs)
def _create_dbus_service(self):
venusversion, venusbuildtime = self._get_venus_versioninfo()