Callback for dataclass or configuration file updates

[yateya]

Hi,

Is there any way to run some code when object is updated after the yml file is updated?

In this example code, I want to reconnect to correct broker when broker address is modified in the yml file.

I found a work-around by checking _last_load attribute; which seems not a good option.

from datafiles import datafile
import time
import threading
from paho.mqtt.client import Client
import random

def get_temp_c():
    return random.randint(0, 100)

def get_fan():
    return random.randint(0, 3)

def compare_fans(fan1, fan2):
    return fan1 != fan2

@datafile('/tmp/broker.yml', manual=False)
class Broker:
    mqtt_addr: str = 'localhost'
    temp_topic: str = 'device/temp_c'
    fan_topic: str = 'device/fan'

broker = Broker()
print(broker)

mqtt_client = Client()
mqtt_client.connect(Broker.mqtt_addr)

def watch_temp():
    last_temp = 0
    while True:
        temp_c = get_temp_c()
        print("Temp: ", temp_c)
        if temp_c != last_temp:
            last_temp = temp_c
            print("Publishing temp: ", temp_c)
            mqtt_client.publish(Broker.temp_topic, temp_c)
        time.sleep(5)

def watch_fan():
    last_fan = 0
    while True:
        fan = get_fan()
        print("Fan: ", fan)
        if compare_fans(fan, last_fan):
            last_fan = fan
            print("Publishing fan: ", last_fan)
            mqtt_client.publish(Broker.fan_topic, last_fan)
        time.sleep(5)

def watch_changes():
    #Work around to watch for file changes
    #Get last load time
    last_load = broker.datafile._last_load
    while True:
        if broker.datafile.modified: #sometimes files are not auto-loaded
            print('broker.yml modified, reloading')
            broker.datafile.load()
        #Check if datafile has been updated, then reconnect to new address
        if last_load != broker.datafile._last_load:
            print('broker.yml updated, reconnecting to new address')
            last_load = broker.datafile._last_load
            mqtt_client.disconnect()
            mqtt_client.connect(Broker.mqtt_addr)
        time.sleep(1)

temp_thread = threading.Thread(target=watch_temp, daemon=True)
temp_thread.start()

fan_thread = threading.Thread(target=watch_fan, daemon=True)
fan_thread.start()

watch_changes_thread = threading.Thread(target=watch_changes)
watch_changes_thread.start()
watch_changes_thread.join()

Having a callback with old/new values will be very helpful. For example like this:

@datafile('broker.yml', mod_callback=values_changed)
class Broker:
    mqtt_addr: str = 'localhost:1883'
    def values_changed(old):
       #compare mqtt_addr with old.mqtt_addr
       if mqtt_addr != old.mqtt_addr:
                   mqtt_client.disconnect()
                  mqtt_client.connect(mqtt_addr)

[jacebrowning]

My first through is to rely on manual synchronization:

@datafile('broker.yml', manual=True)
class Broker:
    mqtt_addr: str = 'localhost:1883'

    def check_mqtt_addr(self):
        if self.datafile.modified:
            old_mqtt_addr = self.mqtt_addr
            self.datafile.load()
            if self.mqtt_addr != old_mqtt_addr:
                mqtt_client.disconnect()
                mqtt_client.connect(self.mqtt_addr)  

broker = Broker()

mqtt_client = Client()
mqtt_client.connect(broker.mqtt_addr)

while True:
    time.sleep(1)
    broker.check_mqtt_addr()

Do you think that would work for your use case?

[yateya]

Thanks for the reply. Yes, it looks good. I will give it a try.