smartHome Zeitschaltuhr 2.0

[der_Mausbiber]

Zu dem Ganzen hätte ich noch eine Frage:

Ist es möglich den den Namen einer Klasse bei deren Initialisierung durch eine Variabel zu ersetzen?

quasi anstatt

Code: Alles auswählen

switches[switches_info[result['switches_id'], "index"]] = BrickletQuadRelay(result['argA'],
                                                                                        result['switches_id'],
                                                                                        DEVICE_IP,
                                                                                        tinkerforge_connection, logger,
                                                                                        consumers)

etwas in dieser Richtung

Code: Alles auswählen

klassenname = result['class']
switches[switches_info[result['switches_id'], "index"]] = klassenname(result['argA'],
                                                                                        result['switches_id'],
                                                                                        DEVICE_IP,
                                                                                        tinkerforge_connection, logger,
                                                                                        consumers)

Dann könnte man die Schalterkonfiguration schon etwas vereinfachen.

[Dav1d]

Ist es möglich den den Namen einer Klasse bei deren Initialisierung durch eine Variabel zu ersetzen?

Ja.

[der_Mausbiber]

Könntest du mir ganz kurz erklären wie?

[Sirius3]

dafür nimmt man ein Wörterbuch:

Code: Alles auswählen

NAME2SWITCHCLASS = {
    "BrickletQuadRelay": BrickletQuadRelay,
    ...
}

switchclass = NAME2SWITCHCLASS[klassenname]

[der_Mausbiber]

Okay, das klingt einfach.

Kann ich das Wörterbuch auch aus einer Datenbank generieren oder muss ich die Daten im Wörterbuch manuell eingeben?

[Sirius3]

@der_Mausbiber: im Wörterbuch stehen die Pythonklassen, die Du ja instanziieren willst. Das geht nicht über eine Datenbank.

[der_Mausbiber]

Hmm, okay, dann kann ich den Code damit zwar vereinfachen, ein manueller Eingriff ist aber immer noch notwendig.
Schade.

Ich hoffe das "__deets__" nochmal kurz Zeit findet und mir sagen kann ob ich seine Idee richtig verstanden habe.

[__deets__]

Hallo,

zuerst einmal klingt das so, als ob du meine Grundidee verstanden hast. Zu deinen Fragen ein paar Antworten, und weil ja (voellig zu Recht) auf erweiterte Schalter hinweist, versuche ich auch da etwas zu zu sagen.

Zuerst einmal ist es prinzipiell ja kein Problem, meinen Ansatz auf mehrere Schalter zu generalisieren. Ein Client meldet dann eben immer eine ganze Reihe von Schaltern an, und ein Schalter ist damit auch abgedeckt. Jeder Schalter bekommt einen Namen, und der muss eindeutig sein. Und diese Forderung ist auch denke ich simpel genug zu erfuellen fuer die Benutzer. Wer dabei einen Fehler macht, macht das auch anderweitig.

Und genau dieser Name ist auch die ID. Sobald ein neuer Schalter vom einem der Clients (oder einem neuen) gemeldet wird, steht der zur Verfuegung. Vorher kann man da ja eh nicht wirklich etwas mit anfangen, ich denke mal nicht, dass Leute sich komplexe Schaltszenarien ausdenken und einpflegen, bevor sie einen Schalter wirklich verbunden und konfiguriert haben.

Bestenfalls braucht die Oberflaeche dann einen Weg, einen bekannten Schalter rauszuwerfen, wenn man weiss, dass der nicht mehr existiert.

Die Frage, ob die Konfiguration uebersichtlich ist oder nicht ist schwieriger zu beantworten. Zuerst einmal sollte man sich vergegenwaertigen, dass die Einrichtung ein seltenes Ereignis ist. Wenn ich also nach der "Pareto-Regel" vorgehe, und versuche mit so wenig Aufwand wie moeglich so viel Ergebnis wie moeglich zu erzielen, wuerde ich immer den Wert eher auf die leichte Bedienbarkeit eines konfigurierten Systems legen, statt die Konfiguration so einfach wie moeglich zu machen.

Dann ist natuerlich das entwanzen eines komplexen, interagierenden Systems wie deiner Client-Server-Anwendung schwieriger, als eine Komponente durchzutesten, bis man sich sicher ist, dass sie funktioniert.

Ich wuerde zB den Client immer so bauen, dass er die Konfiguration nimmt, und auch gleich eine Test-Schaltung vornehmen kann, also so etwas:

Code: Alles auswählen

$ gpio-client --config config.json --test-mode cyclic-toggle --test-switch garagenlicht

Dann kannst du dort pruefen, ob alles geht, ohne dass du den Server ueberhaupt bemuehen musst.

[__deets__]

Ich sehe jetzt erst, das in der Zwischenzeit noch Nachrichten reingekommen sind.

Woher soll denn die Datenbank fuer die Klassennzuordnung kommen? Das waere ja wieder Spezialcode fuer die Schalter im Server, der davon gar nichts wissen soll. Der kennt nur Schaltername und Zustand, und die Zeitablaufssteuerung. Oder eine DB-Verbindung im Client, die nicht noetig ist

Auf *clientseite* kannst du das natuerlich ggf. machen mit dem Woerterbuch (handgeschrieben), ich weiss aber nicht, ob es nicht sinnvoller ist, fuer jeweils eine Sorte Schalter einen Client zu schreiben. Ob man jetzt einen oder drei Prozesse mit Clients hat macht keinen grossen Unterschied, aber die Komplexitaet im Code scheint mir im Moment eher dein Problem - loes die Dinge einzeln, und dann laesst sich auch besser sehen, was verallgemeinerbar ist und was nicht.

Was mir auch noch einfaellt: ich kenne das Websocket-Modul nicht. Aber ich seher auch ersteinmal kein Problem darin, das jeder Client die Schaltnachrichten der anderen bekommt. Das macht die Programmierung einfacher, und die Last an Schaltnachrichten ist ja sehr gering - aus Sicht eines Computers dauert es Aeonen zwischen den Schaltvorgaengen, wenn wir Menschen schon mit einem epileptischen Anfall am Boden liegen. Da muss man nicht optimieren, solange alles im selben Netzwerk haengt.

[__deets__]

So, hier mal eine kleine Skizze:

Code: Alles auswählen

from __future__ import print_function

import sys
import json
import argparse
import time


def error(message, code=1):
    sys.stderr.write(message)
    sys.stderr.write("\n")
    sys.exit(code)


class Switch(object):

    def __init__(self, config, on_off_function):
        self._name = config["name"]
        self._inverted = config["invert"]
        self._on = False
        self._on_off_function = on_off_function
        self.set(False)


    def set(self, state):
        self._on = state
        real_state = state ^ self._inverted
        self._on_off_function(real_state)



class SwitchClientBase(object):

    def __init__(self):
        self._switches = {}


    def add_specific_arguments(self, parser):
        """
        Overload this to add specific arguments for your switch implementation
        """
        pass


    def run(self):
        parser = argparse.ArgumentParser()
        parser.add_argument("--config", help="configuration file")
        parser.add_argument("--print-basic-config", action="store_true")
        parser.add_argument("--test-mode", choices=["cyclic", "on", "off"])
        self.add_specific_arguments(parser)

        opts = parser.parse_args()
        if opts.print_basic_config:
            print("A basic configuration. All keys with `_description` are just explanations and can be left out")
            print(json.dumps(self.basic_config(), indent=4))
            sys.exit(0)

        if not opts.config:
            error("You need to give a config-file. Get an example using the `--print-basic-config` argument")

        with open(opts.config) as inf:
            config = json.load(inf)


        for switch in config["switches"]:
            self._switches[switch["name"]] = self._setup_switch(switch)


        if opts.test_mode is not None:
            getattr(self, "test_%s" % opts.test_mode)(opts)


    def test_cyclic(self, opts):
        state = True
        while True:
            time.sleep(1.0)
            for switch in self._switches.values():
                switch.set(state)
            state = not state


    def basic_config(self):
        return {
            "uri" : "server URI such as tcp://localhost:12345",
            "switches" : [
                {
                    "name": "unique switch name, e.g garage",
                    "invert" : False,
                    "invert_description" : "If True, the switch state is inverted, meaning a ON switch will deliver a LOW-level",
                },
                {
                    "name": "unique switch name, e.g terrarium",
                    "invert" : True,
                    "invert_description" : "If True, the switch state is inverted, meaning a ON switch will deliver a LOW-level",
                },
            ],
        }


    def _setup_switch(self, config):
        on_off_function = self.on_off_function(config)
        return Switch(config, on_off_function)



    def on_off_function(self, config):
        raise NotImplementedError



class GPIOClient(SwitchClientBase):


    def basic_config(self):
        config = super(GPIOClient, self).basic_config()
        # augment with our specific info
        for pin, switch in enumerate(config["switches"]):
            switch["gpio_pin"] = pin
            switch["gpio_pin_description"] = "Number of the GPIO-pin to toggle for that switch, in Broadcom-Numbering"
        return config


    def on_off_function(self, config):
        # in real, we would set up the GPIO configured here
        pin = config["gpio_pin"]
        def on_off_function(state):
            print("GPIO", pin, state)

        return on_off_function



def main():
    client = GPIOClient()
    client.run()


if __name__ == '__main__':
    main()

Mit Konfiguration:

Code: Alles auswählen

{
    "switches": [
        {
            "invert": false,
            "name": "garage",
            "gpio_pin": 0
        },
        {
            "invert": true,
            "name": "terrarium",
            "gpio_pin": 1
        }
    ],
    "uri": "tcp://localhost:12345"
}

Aufgerufen mit

Code: Alles auswählen

$ python /tmp/test-gpio-client.py --config /tmp/test-config.json --test-mode cyclic

bekomme ich dann halt Ausgaben im Sekundenrhytmus.

Achtung, Server-Verbindung oder andere Testmodi sind noch nicht implementiert!

[__deets__]

So, jetzt ein Server:

Code: Alles auswählen

from __future__ import print_function

import sys
import nanomsg
import json
import threading
import time, random
from functools import partial

REGISTERED_SWITCHES = {}


def random_switching(socket):
    while True:
        time.sleep(1.0)
        if REGISTERED_SWITCHES:
            name = random.choice(REGISTERED_SWITCHES.keys())
            guid = REGISTERED_SWITCHES[name]
            state = random.random() > .5
            message = json.dumps(
                {
                    "switches" : [ { "name" : name, "state" : state }],
                    "guid": guid,
                }
                )
            print(message)
            socket.send(message)


def register(message):
    guid = message["guid"]
    name = message["name"]
    REGISTERED_SWITCHES[name] = guid


def main():
    socket = nanomsg.Socket(nanomsg.BUS)
    socket.bind("ipc:///tmp/test")

    t = threading.Thread(target=partial(random_switching, socket))
    t.daemon = True
    t.start()


    while True:
        message = json.loads(socket.recv())
        print(message)
        register(message)



if __name__ == '__main__':
    main()

Ich benutze nanomsg, weil ich das gut kenne - das sollte aber auch mit Websockets klappen, das nanomsg.BUS-Protokoll hat naemlich dieselbe Eigenschaft die du beschreibst: es sind immer broadcast-Nachrichten an alle angeschlossenen Clients. Darum mache ich die durch die GUID eindeutig - obwohl das strenggenommen gar nicht notwendig waere, denn die namen muessen das ja auch sein.

Und hier der neue Client, den man auch mehrfach starten kann mit unterschiedlichen Konfigurationen:

Code: Alles auswählen

from __future__ import print_function

import sys
import json
import argparse
import time
import uuid
import nanomsg


def error(message, code=1):
    sys.stderr.write(message)
    sys.stderr.write("\n")
    sys.exit(code)


class Switch(object):

    def __init__(self, config, on_off_function):
        self._name = config["name"]
        self._inverted = config["invert"]
        self._on = False
        self._on_off_function = on_off_function
        self.set(False)


    def set(self, state):
        self._on = state
        real_state = state ^ self._inverted
        self._on_off_function(real_state)



class SwitchClientBase(object):

    def __init__(self):
        self._switches = {}
        self._socket = None
        self._guid = str(uuid.uuid4())


    def add_specific_arguments(self, parser):
        """
        Overload this to add specific arguments for your switch implementation
        """
        pass


    def run(self):
        parser = argparse.ArgumentParser()
        parser.add_argument("--config", help="configuration file")
        parser.add_argument("--print-basic-config", action="store_true")
        parser.add_argument("--test-mode", choices=["cyclic", "on", "off"])
        self.add_specific_arguments(parser)

        opts = parser.parse_args()
        if opts.print_basic_config:
            print("A basic configuration. All keys with `_description` are just explanations and can be left out")
            print(json.dumps(self.basic_config(), indent=4))
            sys.exit(0)

        if not opts.config:
            error("You need to give a config-file. Get an example using the `--print-basic-config` argument")

        with open(opts.config) as inf:
            config = json.load(inf)


        for switch in config["switches"]:
            self._switches[switch["name"]] = self._setup_switch(switch)


        if opts.test_mode is not None:
            getattr(self, "test_%s" % opts.test_mode)(opts)
        else:
            self._socket = nanomsg.Socket(nanomsg.BUS)
            self._socket.connect(config["uri"])
            self._register_switches()
            while True:
                message = json.loads(self._socket.recv())
                if message["guid"] == self._guid:
                    self._dispatch(message)


    def _dispatch(self, message):
        if "switches" in message:
            for switch in message["switches"]:
                self._switches[switch["name"]].set(switch["state"])


    def _register_switches(self):
        for name in self._switches:
            self._socket.send(
                json.dumps(
                    { "guid" : self._guid, "name" : name}
                )
            )


    def test_cyclic(self, opts):
        state = True
        while True:
            time.sleep(1.0)
            for switch in self._switches.values():
                switch.set(state)
            state = not state


    def basic_config(self):
        return {
            "uri" : "server URI such as tcp://localhost:12345",
            "switches" : [
                {
                    "name": "unique switch name, e.g garage",
                    "invert" : False,
                    "invert_description" : "If True, the switch state is inverted, meaning a ON switch will deliver a LOW-level",
                },
                {
                    "name": "unique switch name, e.g terrarium",
                    "invert" : True,
                    "invert_description" : "If True, the switch state is inverted, meaning a ON switch will deliver a LOW-level",
                },
            ],
        }


    def _setup_switch(self, config):
        on_off_function = self.on_off_function(config)
        return Switch(config, on_off_function)



    def on_off_function(self, config):
        raise NotImplementedError



class GPIOClient(SwitchClientBase):


    def basic_config(self):
        config = super(GPIOClient, self).basic_config()
        # augment with our specific info
        for pin, switch in enumerate(config["switches"]):
            switch["gpio_pin"] = pin
            switch["gpio_pin_description"] = "Number of the GPIO-pin to toggle for that switch, in Broadcom-Numbering"
        return config


    def on_off_function(self, config):
        # in real, we would set up the GPIO configured here
        pin = config["gpio_pin"]
        name = config["name"]
        def on_off_function(state):
            print("GPIO", name, pin, state)

        return on_off_function



def main():
    client = GPIOClient()
    client.run()


if __name__ == '__main__':
    main()

[der_Mausbiber]

Vielen Dank für die ausführliche Antwort.
Ich muss den Stoff jetzt erstmal in Ruhe durchgehen ...