Chaospott/powerctl
43
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Re-subscribe on reconnect

Browse Source
This commit is contained in:
T 2025-01-27 17:13:30 +01:00
parent e8395ae404
commit 247d7712b9
2 changed files with 17 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
power_mqtt1.py
View File

@ -153,9 +153,12 @@ class PowerCTL:
def mqtt_send_event(self, relay: ButtonInput, ev): def mqtt_send_event(self, relay: ButtonInput, ev):
self.mqtc.publish(f"{self.PREFIX}/{relay.name}/input", ev, qos=0, retain=False) self.mqtc.publish(f"{self.PREFIX}/{relay.name}/input", ev, qos=0, retain=False)
@staticmethod def on_connect(self, _mosq, _obj, connect_flags, _rc):
def on_connect(_mosq, _obj, _foo, _bar): print("Connected", connect_flags)
print("Connected") if not connect_flags.get("session present", 0):
print("Re-Subscribe")
for topic in self.relays:
self.mqtc.subscribe(topic, 0)
@staticmethod @staticmethod
def on_subscribe(_mosq, _obj, _mid, _granted_qos): def on_subscribe(_mosq, _obj, _mid, _granted_qos):

17
test.py
View File

@ -9,21 +9,26 @@ import power_mqtt1
class SMBus: class SMBus:
def __init__(self, _bus): def __init__(self, _bus):
pass self.storage = {}
def read_byte(self, _i2c_addr): def read_byte(self, _i2c_addr):
time.sleep(5) time.sleep(5)
return random.randbytes(1)[0] # return random.randbytes(1)[0]
return self.storage.get(_i2c_addr, -1)
def write_byte(self, i2c_addr, data): def write_byte(self, i2c_addr, data):
pass self.storage[i2c_addr] = data
class MqttClient(mqtt.Client): class MqttClient(mqtt.Client):
def connect(self, host, *args, **kwargs): def connect(self, host, *args, **kwargs):
return super().connect("127.0.0.1", *args, **kwargs) return super().connect("127.0.0.1", *args, **kwargs)
# return super().connect(host, *args, **kwargs)
def publish(self, topic, payload=None, qos=0, retain=False, properties=None, ):
pass
with patch('smbus.SMBus', new=SMBus) as SMBus_mock, \ with patch('smbus.SMBus', new=SMBus) as SMBus_mock:
patch("paho.mqtt.client.Client", new=MqttClient) as connect_mock: with patch("paho.mqtt.client.Client", new=MqttClient) as connect_mock:
power_mqtt1.main() power_mqtt1.main()