kunsi/sdm630_mqtt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

make stuff configurable

Browse source
This commit is contained in:
Franzi 2024-07-30 18:32:03 +02:00
parent 6d49fca981
commit 778cefa9b1
Signed by: kunsi
GPG key ID: 12E3D2136B818350
2 changed files with 32 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
printout.py
View file

@ -1,5 +1,7 @@
import logging
from sys import argv, exit
from time import sleep
from tomllib import load
import paho.mqtt.client as mqtt
from rich.align import Align
@ -8,9 +10,15 @@ from rich.live import Live
from rich.panel import Panel
from rich.table import Table
MQTT_HOST = "127.0.0.1"
MQTT_SUB = "sdm630/"
TITLE = "SDM630"
try:
with open(argv[1], "rb") as f:
config = load(f)
TITLE = config["printout"]["title"]
MQTT_SUB = "{}/#".format(config["mqtt"]["prefix"])
except Exception as e:
print(f"Usage: {argv[0]} config.toml")
exit(1)
TABLE_LAYOUT = {
"row1": {
"voltage": {
@ -50,7 +58,7 @@ mqtt_data = {}
def on_connect(client, userdata, flags, rc):
LOG.info(f"Connected to mqtt server")
mqtt.subscribe(f"{MQTT_SUB}#")
mqtt.subscribe(MQTT_SUB)
def on_disconnect(client, userdata, rc):
@ -98,7 +106,7 @@ try:
mqtt.on_connect = on_connect
mqtt.on_disconnect = on_disconnect
mqtt.on_message = on_message
mqtt.connect(MQTT_HOST, 1883, 10)
mqtt.connect(config["mqtt"]["host"], config["mqtt"]["port"], 10)
mqtt.loop_start()
with Live(generate_layout(), screen=True) as live:

37
sdm630_mqtt.py
View file

@ -1,6 +1,7 @@
import logging
from sys import argv
from sys import argv, exit
from time import sleep
from tomllib import load
import paho.mqtt.client as mqtt
from pyModbusTCP.client import ModbusClient
@ -9,27 +10,23 @@ from pyModbusTCP.utils import decode_ieee, word_list_to_long
logging.basicConfig(level=logging.DEBUG)
LOG = logging.getLogger(__name__)
class SDM630_MQTT:
def __init__(
self,
modbus_host,
mqtt_host="localhost",
modbus_port=4196,
mqtt_port=1883,
modbus_unit=1,
mqtt_prefix="sdm630",
config,
):
self.modbus = ModbusClient(
host=modbus_host,
port=modbus_port,
unit_id=modbus_unit,
host=config["modbus"]["host"],
port=config["modbus"]["port"],
unit_id=config["modbus"]["unit_id"],
)
self.modbus.timeout = 10
self.mqtt = mqtt.Client()
self.mqtt_host = mqtt_host
self.mqtt_port = mqtt_port
self.mqtt_prefix = mqtt_prefix
self.mqtt_host = config["mqtt"]["host"]
self.mqtt_port = config["mqtt"]["port"]
self.mqtt_prefix = config["mqtt"]["prefix"]
self.mqtt.on_connect = self._on_mqtt_connect
self.mqtt.on_disconnect = self._on_mqtt_disconnect
@ -120,14 +117,18 @@ class SDM630_MQTT:
if __name__ == "__main__":
client = SDM630_MQTT(
"192.168.1.254",
"127.0.0.1",
)
try:
with open(argv[1], "rb") as f:
config = load(f)
TITLE = config["printout"]["title"]
except Exception as e:
print(f"Usage: {argv[0]} config.toml")
exit(1)
client = SDM630_MQTT(config)
client.start()
try:
while True:
client.request_and_publish()
except Exception as e:
LOG.exception('oops')
LOG.exception("oops")
client.stop()