smarthome/powerplug-energenie
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

validate and publish always with all channels; simplifications

Browse Source
This commit is contained in:
Dirk Alders 2022-07-24 10:38:30 +02:00
parent c796d1d47a
commit 1ccc230b50
1 changed files with 38 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
powerplug.py
View File

@ -26,35 +26,48 @@ class sispmctl(object):
logger.error("sispm error code %d", grepexc.returncode)
else:
logger.info('sispmctl all channels switched off')
self.publish_states()
def __filter_output_parameter__(self, output):
try:
return int(output)
except ValueError:
return output
def set_out_state(self, output, state):
if output in range(1,5):
if self.get_out_state(output) != state:
try:
out_txt = subprocess.check_output(["sudo", "sispmctl", "-o" if state else "-f", str(output)]).decode('UTF-8')
except subprocess.CalledProcessError as grepexc:
logger.error("sispm error code %d", grepexc.returncode)
else:
logger.info('sispmctl channel %d changed to %s', output, state)
cnt = 0
while cnt < 3 and self.__state__[output - 1] != state:
cnt += 1
try:
out_txt = subprocess.check_output(["sudo", "sispmctl", "-g", str(output)])
except subprocess.CalledProcessError as grepexc:
logger.error("sispm error code %d", grepexc.returncode)
else:
self.__state__[output - 1] = out_txt.decode("UTF-8").split("\n")[1].split("\t")[1] == "on"
time.sleep(.2)
self.send_state(output)
if output == "all":
state = [state, state, state, state]
if self.get_out_state(output) != state:
try:
out_txt = subprocess.check_output(["sudo", "sispmctl", "-o" if state == True or state == [True, True, True, True] else "-f", output]).decode('UTF-8')
except subprocess.CalledProcessError as grepexc:
logger.error("sispm error code %d", grepexc.returncode)
else:
logger.info('sispmctl channel %s changed to %s', output, state)
self.publish_states()
def publish_states(self):
try:
out_txt = subprocess.check_output(["sudo", "sispmctl", "-g", "all"])
except subprocess.CalledProcessError as grepexc:
logger.error("sispm error code %d", grepexc.returncode)
else:
print(out_txt)
for i in range(1, 5):
self.__state__[i-1] = out_txt.decode("UTF-8").split("\n")[i].split("\t")[1] == "on"
self.send_state(i)
def send_state(self, output):
topic = config.MQTT_TOPIC + "/status/" + str(output)
logger.info("Sending Powerplug status information of plug %d to mqtt %s = %s", output, topic, str(self.get_out_state(output)))
logger.info("Sending Powerplug status information of plug %s to mqtt %s = %s", str(output), topic, str(self.get_out_state(output)))
self.__mqtt_client__.publish(topic, "true" if self.get_out_state(output) else "false")
def get_out_state(self, output):
return self.__state__[output - 1]
if output in range(1,5):
return self.__state__[int(output) - 1]
else:
return self.__state__
class mqtt_powerplug(object):
@ -62,7 +75,8 @@ class mqtt_powerplug(object):
"set/1",
"set/2",
"set/3",
"set/4"
"set/4",
"set/all"
]
def __init__(self):
@ -79,9 +93,9 @@ class mqtt_powerplug(object):
def __receive__(self, client, userdata, message):
if message.topic in self.__topics__:
output = int(message.topic.split("/")[-1])
output = message.topic.split("/")[-1]
state = message.payload == b"true"
logger.info("Received request to set output channel %d to state %s", output, str(state))
logger.info("Received request to set output channel %s to state %s", output, str(state))
self.__sc__.set_out_state(output, state)
else:
logger.warning("Ignoring unknown mqtt topic %s", message.topic)