Module¶

Mjukvarumoduler skapas som Python-klasser. Modulen laddas därefter genom att lägga till den i konfigurationsfilen.

Definiera en modul¶

En modul är en Pythonklass som antingen ärver ifrån DecoratedModule (i särskilda fall kan Module ärvas från istället). I exemplet nedan så antas filen ligga i lrpy/modules/test_mod.py.

from lrpy import DecoratedModule


class TestModule(DecoratedModule):
    pass

Genom konfigurationen nedan kan modulen laddas.

  commod:
  type: "lrpy.com_module:StdComModule" # Ladda kom-modulen (bör alltid vara samma)
  config:
    modules: # Lista av moduler som kom-modulen ska ladda
      - type: "lrpy.modules.test_mod:TestModule" # Vår modul!

Type strängen är på formatet fil:klassnamn där fil skrivs som om man skulle importera den (på samma sätt som from fil import klassnamn).

Ta emot meddelanden¶

Om DecoratedModule används så används @on_message för att markera att en funktion lyssnar efter ett meddelande. Funktionen ska ta self och message, där message har en type hint som anger vilken meddelandetyp funktionen lyssnar på (dvs message: TestMessage lyssnar på TestMessage). Nedan visas ett exempel på hur en modul kan lyssna på ett meddelande.

from lrpy import DecoratedModule, on_message
from lrpy.message import TestMessage


class TestModule(DecoratedModule):
    @on_message
    async def on_test(self, message: TestMessage):
        print(message.value)

Flera meddelanden kan lyssnas på i samma funktion enligt nedan

@on_message
async def on_test(self, message: TestMessage | TestMessage2):
    print(message)

Skicka ett meddelande¶

import asyncio

from lrpy import DecoratedModule
from lrpy.message import TestMessage


class TestModule(DecoratedModule):
    async def run(self, stop: asyncio.Event):
        while not stop.is_set():
            self.emit_message(
                TestMessage(value=1)
                # Man måste använda kwargs (d.v.s. value=5 istället för 5) när man skapar meddelanden
            )

Alla meddelandetyper är definierade i filen lrpy/message.py

Run & huvudloop för modulen¶

En modul kan köra kod kontinuerligt genom att definiera run-metoden. Metoden tar emot ett stop-event som signalerar när programmet ska avslutas (t.ex. när användaren trycker Ctrl+C).

Modulen måste respektera stop-eventet och avsluta inom ett par sekunder efter att det satts. Det innebär:

Använd while not stop.is_set() som loop-villkor.
Använd await asyncio.sleep() med ett rimligt intervall (högst ett par sekunder) så att loopen hinner kontrollera stop regelbundet.
Om modulen gör blockerande arbete (t.ex. väntar på I/O) måste den ändå avbryta inom rimlig tid.

import asyncio

from lrpy import DecoratedModule
from lrpy.message import TestMessage


class TestModule(DecoratedModule):
    async def run(self, stop: asyncio.Event):
        while not stop.is_set():
            self.emit_message(
                TestMessage(value=1)
            )
            await asyncio.sleep(1)  # Vänta 1 sekund — loopen avslutas inom ~1s efter stop

await asyncio.sleep() fyller två syften:

Den ger andra moduler och event-loopen tid att köra.
Den begränsar hur lång tid det tar innan loopen märker att stop har satts.

Om en modul inte avslutas inom 10 sekunder efter att stop satts kommer dess async-task att termineras (via asyncio.Task.cancel) och ett felmeddelande skrivs ut. Detta är en säkerhetsmekanism — moduler ska inte förlita sig på den utan istället avsluta ordentligt på egen hand. En sleep på t.ex. 60 sekunder skulle alltså leda till att modulen tvingas att avslutas varje gång programmet stoppas.

Om modulen inte behöver en loop (t.ex. om den bara lyssnar på meddelanden) behöver run inte definieras.

Telemetri¶

Moduler kan exponera telemetridata som visas i frontenden. Telemetri definieras som en lista av TelemetryField i klassattributet TELEMETRY.

import asyncio

from lrpy import DecoratedModule
from lrpy.telemetry import TelemetryField, TelemetryState


class TestModule(DecoratedModule):
    TELEMETRY = [
        TelemetryField("temperature", "Temperatur", "Aktuell temperatur", unit="°C"),
        TelemetryField("status", "Status", "Modulens status"),
    ]

    async def run(self, stop: asyncio.Event):
        while not stop.is_set():
            self.telemetry.temperature.set(42, state=TelemetryState.OK)
            self.telemetry.status.set("Kör", state=TelemetryState.OK, text="Allt fungerar")
            await asyncio.sleep(1)

TelemetryField tar följande argument:

id — unikt id som används för att komma åt värdet via self.telemetry.<id>
name — visningsnamn
description (valfri) — beskrivning
unit (valfri) — enhet, t.ex. "ms", "°C"

Värden uppdateras med self.telemetry.<id>.set(value, state, text) där state anger färg/status:

TelemetryState.NEUTRAL — standard
TelemetryState.OK — grönt/ok
TelemetryState.WARN — varning
TelemetryState.ERROR — fel

text är ett valfritt meddelande som ger mer kontext om värdet.

Loggning¶

Varje modul har en inbyggd logger (self.logger) som kan användas för att skriva loggmeddelanden. Loggarna visas i webbgränssnittet under systempanelen.

Det finns fyra loggnivåer, från minst till mest allvarlig:

Metod	Användning
`self.logger.debug()`	Visas oftast inte, är till för specifik felsökning
`self.logger.info()`	Beskriver vad som händer och ger kontext till resten av loggen
`self.logger.warning()`	Indikerar potentiella problem som inte hindrar korrekt funktion
`self.logger.error()`	Något fungerar inte

Undvik att logga onödigt mycket. Det spammar loggen och gör den svår att använda samt påverkar prestandan negativt. Logga istället vid händelser som tillståndsförändringar eller fel. Telemetrisystemet kan oftast används istället.

Exempel:

from lrpy import DecoratedModule


class TestModule(DecoratedModule):
    async def run(self, stop):
        self.logger.info("Modulen startar")
        sensor_ok = True

        while not stop.is_set():
            data = self.read_sensor()

            if data is None and sensor_ok:
                self.logger.warning("Fick ingen data från sensorn")
                sensor_ok = False
            elif data is not None and not sensor_ok:
                self.logger.info("Sensorn svarar igen")
                sensor_ok = True

            await asyncio.sleep(1)

Loggern sätts automatiskt upp med modulens klassnamn, så det syns i loggen vilken modul som skrev meddelandet.

Konfiguration¶

En modul kan ta emot konfiguration från YAML-filen genom att definiera en inre Config-klass som ärver från pydantic.BaseModel.

from pydantic import BaseModel, Field

from lrpy import DecoratedModule


class TestModule(DecoratedModule):
    class Config(BaseModel):
        interval: float = 1.0
        device_path: str = Field(alias="device-path")

    def __init__(self, config: Config):
        self.config: TestModule.Config = config
        super().__init__(config)

Konfigurationen skickas in via YAML-filen:

modules:
  - type: "lrpy.modules.my_module:TestModule"
    config:
      interval: 0.5
      device-path: "/dev/i2c-1"

Config-klassen valideras automatiskt av Pydantic vid laddning. Om ett fält i YAML-filen använder bindestreck (t.ex. device-path) kan Field(alias="device-path") användas för att mappa det till ett Python-attribut (t.ex. device_path).

Om modulen inte definierar en Config-klass behöver inget config-block anges i YAML-filen.