23  Edge System

23.1 Einführung

Programmieren lernt man bekanntlich nur durch programmieren. Angelehnt an die aktuellen Themen im IT/OT Bereich, bietet sich doch dabei ein EdgeSystem für Sensoren an.

Unser Anwendung “EdgeSensors”: Mit EdgeSensors können Messdaten von einem spezifischen Sensor erfasst werden. Der Sensor entstammt aus einer vordefinierten Auswahl und der Benutzer hat die Möglichkeit Intervall und Dauer der Messung zu konfigurieren. Nach der Messung werden die einzelnen Messwerte und Metadaten wie Mittelwert, Min und Max ausgegeben.

Detaillierter Funktionsumfang: - Es kann ein Sensor aus einer vordefinierten Sammlung ausgewählt werden - Die Anzahl Messwerte werden durch Vorgabe des Intervalls und Messdauer definiert - Die Anwendung definiert ein Standard Interface, so dass jeder Sensortyp die gleichen Ein- und Ausgangsparameter hat. Dies erlaubt eine einfache Einbindung in übergeordnete Systeme. - Es können beliebig viele Sensoren implementiert und angesprochen werdne - Die Konfiguration des Tools erfolgt per CLI (Command Line Interface)

Erweiterung: In der Erweiterung kann kann EdgeSensors als Prozess gestartet werden, um kontinuierlich Messdaten zu erfassen. - Starten als Prozess - Messen mit mehreren Sensoren gleichzeitig - Interface für externe Einbindung per API

Was möchten wir lernen - Aufbau einzelner Sensoren - Organisation in unabhängigen Python-Modulen - Aufbau einer Main-Anwendung - Organisation als Klassen zur einheitlichen Ansteuerung - Integrieren eines zusätzlichen Interfaces

23.1.1 Arbeitsumgebung

Die vorgestellten Beispiele werden in Visual Studio Code umgesetzt.

23.2 Projektsetup

Für dieses Projekt wird folgendes Setting eingerichtet:

  1. Einen neuen Projektordner anlegen (z.B. sensor_edge)
  2. Einen Unterordner “sensors”
  3. Im Hauptverzeichnis eine app.py
  4. requirements.txt für die benutzten Python Module
sensor_edge/
├── sensors/
│   └── ...
|   app.py
|   .gitignore
└── requirements.txt

23.3 Erste Sensoren

Für ein erstes Gefühl möchten wir einen einfachen Sensor erstellen. Da das System im Ausblick für mehrere Sensoren gedacht ist, packen wir diesen bereits in ein eigenes Modul (Eigenes *.py File). Im Ordner “sensors” legen wir dafür das Modul rand_sensor.py an.

Im ersten Schritt bietet dieser eine “read_rand_sensor”-Methode an für einen statischen Wert

def read_rand_sensor():
    return 42

Für ein iteratives Vorgehen, möchten wir direkt Testen und ergänzen dafür die Prüfung für __main__ im Modul. Durch diese Condition wird der darin enthaltene Code nur ausgeführt, wenn dieses Modul direkt gestartet anstatt in einem anderen Modul importiert wird.

if __name__ == "__main__":
    value = read_rand_sensor()
    print(value)

Für etwas mehr Spannung nutzen wir das Modul der Standardbibliothek “Random” um den return-Wert etwas interessanter zu gestalten.

from random import randint

def read_rand_sensor():
    return randint(1, 50)

Testen wir! Voilà unser erster Sensor…

23.3.1 Battery Sensor

Für etwas mehr Hardwarebezug, bauen wir uns direkt einen zweiten Sensor, der den Ladezustand unseres Akkus auslesen soll. Dafür wird die Python-Library psutil genutzt. Dieser Library erlaubt Zugriff auf Systeminformationen wie CPU-Frequenzen, Akkuzustände, Prozesse uvm.

Erstellen wir wieder im “sensors” Ordner ein neues Modul battery_sensor.py und fügen eine read_battery_sensor hinzu. Ebenfalls direkt mit __main__ um Testen zu können.

def read_battery_sensor():
    import psutil
    value = psutil.sensors_battery().percent
    return value

if __name__ == "__main__":
    value = read_battery_sensor()
    print(value)

Hinweis installieren sie das Modul psutil (Empfohlen in eigener virtueller Umgebung)

pip install psutil

Somit haben wir einen ersten Hardware-Sensor! Gratulation.

23.4 Gemeinsames Sensorinterface

Hinweis: funktionale Programmierung

EdgeSensors soll es erlauben unterschiedliche Sensoren einheitlich anzusprechen und einheitliche Daten als Feedback zu erhalten. Kurz - Sie brauchen das selbe Interface. Nehmen wir mal folgende Grundfunktionen für jeden Sensor an:

  1. Parameterübergabe - Intervall (Abtastungszeit in Sekunden) und Duration (Messdauer in Sekunden)
  2. Messen (read)
  3. Feedback - Sensorname, Messdaten, eingesetzte Parameter, Startzeit, Metadaten (Min, Max, Average …)
  4. Ein Name zur Identifikation

Diese Anforderungen möchten wir in einem Modul base_sensor.py unter sensors als Basisklasse umsetzen:

class BaseSensor:
    def __init__(self, name, intervall, duration):
        """
        Konstruktor der Basis-Klasse
        """
        self._name = name
        self._intervall = intervall
        self._duration = duration

    def read(self):
        """
        Lese einen Messwert
        Hinweis: Dies ist eine "leere" Funktion die in der jeweiligen Sensorklasse 
        implementiert werden muss. Sollte dies vergessen werden, wird nachfolgende
        Exception ausgelöst.
        """
        raise Exception("Muss von einer ableitenden Klasse implementiert werden")

    def run_measurement(self):
        """
        Startet die Messung und blockiert bis diese abgeschlossen ist
        """
        # Implementierung folgt...
        pass

    def get_values(self):
        """
        Liefert die erfassten Messwerte
        """
        # Implementierung folgt
        pass

Abstrahiert läuft die Durchführung einer Messung unabhängig des Sensortyps, gleich ab. Das heisst eine Funktion wie “run_measurement” sollte in der Basisklasse umgesetzt werden können und allgemein gültig sein. Zur Einfachheit werden zwei statische Parameter Duration (Messdauer) und Intervall (Abtastung) definiert. Genauso möchten wir die Startzeit der Messung festhalten. Im folgenden eine Beispiel-Implementierung basierend auf den Libraries time und datetime.

Hinweis: run_measurement nutzt read() - welches später durch die eigentliche Sensorimplementierung vorgegeben (überschrieben) wird.

import datetime
import time

class BaseSensor:
    def __init__(self, name, intervall, duration):
        """
        Konstruktor der Basis-Klasse
        """
        # Lege zwei Variablen an
        self._measurement_time = None
        self._data = None

    def read(self):
        ...

    def run_measurement(self):
        """
        Starte eine Messung
        """
        # Setze den Startzeitpunkt zur Bestimmung der Dauer
        start_time = time.time()
        self.measurement_time = datetime.datetime.now().isoformat()

        # Leere alle bestehenden Messwerte
        self.data = []

        # Messen...
        while (time.time() - start_time) < self.duration:
            value = self.read()
            self.data.append(value)
            time.sleep(self.intervall)

    def get_values(self):
        ...

Die Ausgabe der Messresultate soll für jeden Sensor gleich und einfach weiter zuverwenden sein. Ein modernes Format ist dafür JSON wie folgendes Beispiel zeigt:

json_example = {
    "name" : "mein_sensor",
    "time" : "ein Datum",
    "intervall" : 0.5,
    "duration" : 1,
    "data" : [1,2,3,4]
}

In Python ist es am einfachsten alle Daten in einem dict zu halten - Die Umwandlung in JSON kann danach einfach mit Hilfe der json Library gelöst werden. json.dumps(Data) wandelt das dict einfach in einen JSON String um.

def get_values(self):
    msg = {
        "name" : self.name,
        "time" : self.measurement_time,
        "intervall" : self.intervall,
        "duration" : self.duration,
        "data" : self.data
    }
    return json.dumps(msg)

23.4.1 Testen der Basisklasse

Zur Sicherstellung der Funktionalität bietet sich ein Test an. Als Beispiel wird innerhalb von “base_sensor.py” ein SimpleSensor implementiert, der BaseSensor ableitet. Wichtig dabei ist, die Übergabe der __init__ Parameter zur Erstellung des Objekts. SimpleSensor könnte aber noch mehr Parameter besitzen. Aus diesem Grund könne __init__ von SimpleSensor noch weitere Parameter haben wie z.B. Verbindungsparameter, Filtereinstellungen usw. Ebenfalls wurde read() überschrieben, in diesem Fall einfach durch eine statische Rückgabe 5.

Minimale Klasse:

class SimpleSensor(BaseSensor):
    """
    SimpleSensor to test BaseSensor
    """
    def __init__(self, name, intervall, duration):
        super().__init__(name, intervall, duration)

    def read(self):
        """
        Reads single measurement
        Example as dict 
        """
        value = 5

        return value

Führen wir einen Test durch

    python -m sensors.base_sensor

    # Ausgabe: {"name": "dummy", "time": "2026-03-06T11:09:00.520837", "intervall": 1, "duration": 5, "data": [5, 5, 5, 5, 5]}

23.4.2 Integration des Batteriesensors auf die Basisklasse

Überführen wir den Battery-Sensor ins neue Konstrukt:

from sensors.base_sensor import BaseSensor

class BatterySensor(BaseSensor):
    def __init__(self, name, intervall, duration):
        super().__init__(name, intervall, duration)

    def read(self):
        """
        Reads single measurement
        """
        import psutil
        value = psutil.sensors_battery().percent
        return value


if __name__ == "__main__":

    batt = BatterySensor("batt", 0.5, 3)
    batt.run_measurement()
    print(batt.get_values())

23.4.3 Erstellen von app.py

Als Main-Eintrittspunkt wird app.py verwendet. Die entwickelte BaseSensorklasse, erlaubt es nun in app.py Sensoren unabhängig ihres Types anzusprechen. Lediglich aufgrund eines Konfigurationswertes “sensor_args”, wird die entsprechende Sensorklasse instanziert. Für die eigentliche Messung ist der Typ aber egal, da das Interface das gleiche ist.

    # app.py
    from sensors.battery_sensor import BatterySensor
    from sensors.rand_sensor import RandSensor

    if __name__ == "__main__":
        
        sensor_args = "battery"

        if sensor_args == "battery":
            sensor = BatterySensor("MeinName", 0.5, 2)
        elif sensor_args == "rand":
            sensor = RandSensor("MeinAndererSensor", 1,5)
        
        # Starte eine Messung wenn ein Sensor gefunden wurde
        if sensor:
            sensor.run_measurement()
            print(sensor.get_values())

23.5 Command Line Interface (CLI)

EdgeSensors wird nun durch ein CLI erweitert. Wir möchten also unseren Service mit Argumenten von Extern starten können wie im folgenden Beispiel:

    app.py --sensor battery --name default_name --intervall 0.5 --duration 2

Resultat:

    {"name": "default_name", "time": "2026-03-06T11:17:14.614775", "intervall": 0.5, "duration": 2, "data": [89, 89, 89, 89]}

Python speichert die übergebenen Argumente automatisch in in einer Liste sys.args. Anstatt die Argumente selbst zu parsen wird für die erleichterte Umsetzung das Modul ArgumentParser genutzt. Damit kann per Konfiguration die erwarteten Argumente, Datentyp und Default Werte angegeben werden.

    from sensors.battery_sensor import BatterySensor
    from argparse import ArgumentParser

    if __name__ == "__main__":
        
        # ArgumentParser ist ein Utility um die Argumente aus sys.args \
        # angenehm zu parsen. 
        parser = ArgumentParser()
        parser.add_argument("--sensor", type=str, default="battery")
        parser.add_argument("--name", type=str, default="default_name")
        parser.add_argument("--intervall", type=float, default=0.5)
        parser.add_argument("--duration", type=float, default=2)

        args = parser.parse_args()

        # Basierend auf dem Sensor Argument wird der entsprechende Sensor gesucht 
        if args.sensor == "battery":
            sensor = BatterySensor(args.name, args.intervall, args.duration)
        
        # Starte eine Messung wenn ein Sensor gefunden wurde
        if sensor:
            sensor.run_measurement()
            print(sensor.get_values())