smarthome/powerplug-energenie
1
0
Fork 0
Código Issues Pull requests Versões Wiki Atividade

cyclic state publish via mqtt

Ver código fonte
Esse commit está contido em:
Dirk Alders 2022-07-24 14:08:49 +02:00
commit 587de798c2
1 arquivos alterados com 16 adições e 15 exclusões
Mostrar estatísticas Baixar arquivo de patch Baixar arquivo de diferenças Expandir todos os arquivos Colapsar todos os arquivos

31
powerplug.py
Ver arquivo

@ -35,39 +35,39 @@ class sispmctl(object):
return output
def set_out_state(self, output, state):
output = self.__filter_output_parameter__(output)
if output == "all":
state = [state, state, state, state]
if self.get_out_state(output) != state:
if self.__state__[output - 1] != 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')
out_txt = subprocess.check_output(["sudo", "sispmctl", "-o" if state == True or state == [True, True, True, True] 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 %s changed to %s', output, state)
self.publish_states()
def publish_states(self):
def get_out_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)
rv = []
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)
rv.append(out_txt.decode("UTF-8").split("\n")[i].split("\t")[1] == "on")
return rv
def publish_states(self):
self.__state__ = self.get_out_states()
for i in range(1, 5):
self.send_state(i)
def send_state(self, output):
topic = config.MQTT_TOPIC + "/status/" + str(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):
if output in range(1,5):
return self.__state__[int(output) - 1]
else:
return self.__state__
logger.info("Sending Powerplug status information of plug %s to mqtt %s = %s", str(output), topic, str(self.__state__[output - 1]))
self.__mqtt_client__.publish(topic, "true" if self.__state__[output - 1] else "false")
class mqtt_powerplug(object):
@ -110,4 +110,5 @@ if __name__ == '__main__':
mp = mqtt_powerplug()
while True:
time.sleep(2)
time.sleep(30)
mp.__sc__.publish_states()