Innovenergy_trunk/firmware/opt/innovenergy/relay1/relay1.py

#!/usr/bin/python -u
# coding=utf-8

from python_libs.ie_dbus.dbus_service import DBusService
from python_libs.ie_utils.filters import DebounceFilter
from python_libs.ie_utils.main_loop import run_on_main_loop
from python_libs.ie_utils.mixins import memoize

import logging
import relay_function
import relay_polarity
import test_state

# noinspection PyUnreachableCode
if False:
	from typing import NoReturn, Optional

logging.basicConfig(level=logging.DEBUG)
_log = logging.getLogger(__name__)

VERSION = '1.2.0'
PRODUCT = 'Relay1'

UPDATE_PERIOD_MS = 2000
BATTERY_SERVICE_PREFIX = 'com.victronenergy.battery.'

# settings
DebounceFilterStrength = PRODUCT + '/DebounceFilterTime'
ConnectLoadThreshold = PRODUCT + '/ConnectLoadThreshold'
NominalPowerOfLoad = PRODUCT + '/NominalPowerOfLoad'
Relay1Function = PRODUCT + '/Relay1Function'
Relay1Polarity = PRODUCT + '/Relay1Polarity'

MinimumSocLimit = 'CGwacs/BatteryLife/MinimumSocLimit'

# remote properties
Relay1State = 'system/Relay/1/State'


class Relay1Service(DBusService):

	def __init__(self):

		super(Relay1Service, self).__init__(PRODUCT.lower())

		self.settings.add_setting(path=DebounceFilterStrength, default_value=180,   min=0, max=10000)
		self.settings.add_setting(path=NominalPowerOfLoad,     default_value=10000, min=0, max=100000)  # watts
		self.settings.add_setting(path=ConnectLoadThreshold,   default_value=50000, min=0, max=50000)  # watts
		self.settings.add_setting(path=Relay1Function,         default_value=relay_function.NONE)
		self.settings.add_setting(path=Relay1Polarity,         default_value=relay_polarity.NORMALLY_OPEN)

		self.own_properties.set('/ProductName',         PRODUCT)
		self.own_properties.set('/Mgmt/ProcessName',    __file__)
		self.own_properties.set('/Mgmt/ProcessVersion', VERSION)
		self.own_properties.set('/Mgmt/Connection',     'dbus')
		self.own_properties.set('/ProductId',           PRODUCT)
		self.own_properties.set('/FirmwareVersion',     VERSION)
		self.own_properties.set('/HardwareVersion',     VERSION)
		self.own_properties.set('/Connected',           1)
		self.own_properties.set('/LogicalState',        0)
		self.own_properties.set('/TestRelay1',          -1, writable=True)

		self.debounce_filter = DebounceFilter(False, self.settings.get(DebounceFilterStrength))

	@memoize
	def is_relay1_available(self):
		# type: () -> bool
		return self.remote_properties.exists(Relay1State)

	@property
	def battery_service(self):
		return next((s for s in self.available_services if s.startswith(BATTERY_SERVICE_PREFIX)), None)

	@property
	def grid_power(self):
		# type: () -> float

		l1 = self.remote_properties.get('system/Ac/Grid/L1/Power').value
		l2 = self.remote_properties.get('system/Ac/Grid/L2/Power').value
		l3 = self.remote_properties.get('system/Ac/Grid/L3/Power').value

		l1 = l1 if isinstance(l1, (float, int)) else 0.
		l2 = l2 if isinstance(l2, (float, int)) else 0.
		l3 = l3 if isinstance(l3, (float, int)) else 0.

		return -(l1 + l2 + l3)

	@property
	def test_relay(self):
		# type: () -> Optional[bool]

		test = self.own_properties.get('/TestRelay1').value

		if test == test_state.NO_TEST:
			return None

		return test == test_state.TEST_RELAY_ON

	@property
	def connect_load_threshold(self):
		# type: () -> int
		return self.settings.get(ConnectLoadThreshold)

	@property
	def disconnect_load_threshold(self):
		# type: () -> int
		return self.connect_load_threshold - self.settings.get(NominalPowerOfLoad)

	@property
	def load_relay_target_state(self):
		# type: () -> Optional[bool]

		if self.grid_power > self.connect_load_threshold:
			return True
		if self.grid_power < self.disconnect_load_threshold:
			return False
		else:
			return None  # dead-band

	@property
	def battery_has_alarm(self):
		# type: () -> bool
		return self.remote_properties.get(self.battery_service + '/NumberOfAlarmFlags').value > 0

	@property
	def soc(self):
		# type: () -> float
		return self.remote_properties.get(self.battery_service + '/Soc').value

	# noinspection PyBroadException
	@property
	def min_soc_limit(self):
		# type: () -> int
		try:
			return int(self.settings.get(MinimumSocLimit))
		except Exception:
			return 101  # force alarm when MinimumSocLimit setting is not readable

	@property
	def alarm_relay_target_state(self):
		# type: () -> bool

		if self.battery_service is None:
			_log.debug('no battery available!')
			return True    # force alarm

		_log.debug('soc = ' + str(self.soc))
		_log.debug('min_soc = ' + str(self.min_soc_limit))
		_log.debug('battery_alarm = ' + str(self.battery_has_alarm))

		return self.soc < self.min_soc_limit or self.battery_has_alarm

	@property
	def relay1_target_state(self):
		# type: () -> Optional[bool]
		
		if self.test_relay is not None:
			_log.info('testing ' + self.relay1_function + ' relay')
			return self.test_relay

		_log.debug('relay1_function is ' + self.relay1_function)

		if self.relay1_function == relay_function.ALARM:
			return self.alarm_relay_target_state
		elif self.relay1_function == relay_function.LOAD:
			return self.load_relay_target_state

		return None

	@property
	def relay1_function(self):
		# type: () -> str
		return self.settings.get(Relay1Function)

	@property
	def relay1_polarity(self):
		# type: () -> str
		return self.settings.get(Relay1Polarity)

	def display_state(self, state):
		# type: (bool) -> str

		if self.relay1_function == relay_function.ALARM:
			return 'alarm is on' if state else 'alarm is off'

		if self.relay1_function == relay_function.LOAD:
			return 'load is connected' if state else 'load is disconnected'

		return 'relay is disabled'

	def set_relay1(self, state):
		# type: (bool) -> NoReturn

		changed = self.remote_properties.set(Relay1State, 1 if state else 0)

		if changed or _log.getEffectiveLevel() <= logging.DEBUG:
			_log.info('Relay is now ' + ('energized' if state else 'de-energized'))

	def debounce(self, state):
		# type: (Optional[bool]) -> Optional[bool]

		if state is None:
			return None

		filter_length = int(self.settings.get(DebounceFilterStrength) * 1000 / UPDATE_PERIOD_MS)
		return self.debounce_filter.update(state, filter_length)

	def update(self):

		if self.relay1_function == relay_function.NONE:
			self.set_relay1(False)                # make sure relay is de-energized when it is disabled
			return

		state = self.relay1_target_state          # using tri-state logic, True/False/None
		if state is None:       		          # None acts as a NOP
			return

		if self.test_relay is None:		          # do not debounce when testing
			state = self.debounce(state)

		self.own_properties.set('/LogicalState', state)
		_log.debug(self.display_state(state))

		if self.relay1_polarity == relay_polarity.NORMALLY_CLOSED:
			_log.debug('inverting command because relay is configured to be normally-closed')
			state = not state

		self.set_relay1(state)


def main():

	_log.info('starting ' + __file__)

	with Relay1Service() as service:
		run_on_main_loop(service.update, UPDATE_PERIOD_MS)

	_log.info(__file__ + ' has shut down')


if __name__ == '__main__':
	main()