Initial powerplug implementation

This commit is contained in:
Dirk Alders 2022-07-22 09:14:52 +02:00
parent a3515ef8ba
commit 2c6cf94ce3
6 changed files with 113 additions and 0 deletions

5
.gitignore vendored
View File

@ -1,3 +1,8 @@
# ---> powerplug-energenie
#
config.py
# ---> Python # ---> Python
# Byte-compiled / optimized / DLL files # Byte-compiled / optimized / DLL files
__pycache__/ __pycache__/

3
.gitmodules vendored Normal file
View File

@ -0,0 +1,3 @@
[submodule "report"]
path = report
url = https://git.mount-mockery.de/pylib/report.git

21
config_example/config.py Normal file
View File

@ -0,0 +1,21 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import os
import report
MQTT_SERVER = "multimedia"
MQTT_TOPIC = "hifi/powerplug"
#
# Logging
#
__BASEPATH__ = os.path.abspath(os.path.dirname(__file__))
APP_NAME = "powerplug"
LOGTARGET = 'stdout' # possible choices are: 'logfile' or 'stdout'
LOGLVL = 'DEBUG'
LOGHOST = 'localhost'
LOGPORT = 19996
formatter = report.LONG_FMT

82
powerplug.py Normal file
View File

@ -0,0 +1,82 @@
import config
import logging
import paho.mqtt.client as mqtt
import report
import subprocess
import time
try:
from config import APP_NAME as ROOT_LOGGER_NAME
except ImportError:
ROOT_LOGGER_NAME = 'root'
logger = logging.getLogger(ROOT_LOGGER_NAME).getChild('main')
class sispmctl(object):
AMP_CHANNEL = 0
def __init__(self):
logger.info("Starting sispmctl module...")
self.__state__ = [False, False, False, False]
self.__state_requests__ = [{}, {}, {}, {}]
try:
out_txt = subprocess.check_output(["sudo", "sispmctl", "-f", "all"]).decode('UTF-8')
except subprocess.CalledProcessError as grepexc:
logger.error("sispm error code %d", grepexc.returncode)
else:
logger.info('sispmctl all channels switched off')
def set_out_state(self, output, state):
if output in range(1,5):
if self.__state__[output - 1] != 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:
self.__state__[output - 1] = state
logger.info('sispmctl channel %d changed to %s', output, state)
def get_out_state(self, output):
return self.__state__[output - 1]
class mqtt_powerplug(object):
SUBTOPICS = [
"set/1",
"set/2",
"set/3",
"set/4"
]
def __init__(self):
self.__client__ = mqtt.Client("mqtt_powerplug") # create client object
self.__client__.on_message = self.__receive__ # attach function to callback
self.__client__.connect("192.168.0.131", 1883) # establish connection
self.__client__.loop_start() # start the loop
self.__topics__ = []
for subtopic in self.SUBTOPICS:
self.__topics__.append(config.MQTT_TOPIC + "/" + subtopic)
self.__client__.subscribe(self.__topics__[-1]) # subscibe a topic
self.__sc__ = sispmctl()
def __receive__(self, client, userdata, message):
if message.topic in self.__topics__:
output = int(message.topic.split("/")[-1])
state = message.payload == b"true"
logger.info("Received request to set output channel %d to state %s", output, str(state))
self.__sc__.set_out_state(output, state)
else:
logger.warning("Ignoring unknown mqtt topic %s", message.topic)
def __del__(self):
self.__client__.loop_stop() # stop the loop
if __name__ == '__main__':
report.appLoggingConfigure(config.__BASEPATH__, config.LOGTARGET, ((config.APP_NAME, config.LOGLVL), ), fmt=config.formatter, host=config.LOGHOST, port=config.LOGPORT)
#
mp = mqtt_powerplug()
while True:
time.sleep(2)

1
report Submodule

@ -0,0 +1 @@
Subproject commit 21bac82e0c459ebf6d34783c9249526a657a6bbd

1
requirements.txt Normal file
View File

@ -0,0 +1 @@
paho-mqtt