DataHub – Ditt interface mot MQTT baserade dataspaces
Hur du med ett enda gränssnitt kan koppla upp dig mot dataströmmar, loggar och dashboards som finns sprida på olika dataspaces.
🌍 Vad är ett Dataspace?
Ett dataspace är ett sätt att organisera dataflöden och åtkomst.
Istället för att varje program direkt behöver känna till hur man ansluter till olika serverar, kan man använda ett mellanlager – en tjänst som gör uppkoppling, åtkomst och hantering enklare. Detta mellanlager kallas ofta för en broker eller mäklare på svenska.
En teknisk lösning som passar in i det tänket ganska bra är MQTT som är ett kommunikations protokoll som funnits ganska länge inom IoT området. Den har en central broker och en publish-subscribe model. Där en tjänst eller en sensor kontinuerligt kan publicera data i vad man kan likna ett chattrum för sensorer medan alla som är intresserade av datat kan ansluta sig till chattrummet för att ta del av dataflödet.
Med en liten enkel justering av MQTT kan man skapa en dataspace broker. I MQTT ska en sensor eller tjänst som publicerar data INTE veta om något om de klienter som konsumerar den. Det är nämligen hela iden med MQTT att producenter och konsumenter av data är helt isärkopplade. Det finns dock ett tillfälle då det ändå är ganska bra att veta om ifall det är någon klient som kopplat upp sig.
Säg att vi tex vill ansluta till en chattrum som ger oss en signal på morgonen när solen går och en signal på kvällen är det går ner. Då solen går upp och ner olika tider på olika platser så skulle vi behöva ett oändligt antal chattrum.
mqtt://server.se/solensuppochnergång/umeå
mqtt://server.se/solensuppochnergång/stockholm
mqtt://server.se/solensuppochnergång/malmö
osv...
Men genom att endast skapa chattrum när det efterfrågas av en klient kan vi lösa detta. Chattrummen kan vara utformade som:
mqtt://server.se/solensuppochnergång/latitude/longitude
där varje klient byter ut longitude och latitude mot den plats där den vill veta upp och nergången. I början finns inga chattrum alls men allt eftersom klienter ansluter till tex.
mqtt://server.se/solenssolensuppochnergång/63.8258/20.2630
så skapas de och när klienten koppar ner så upphör chattrummet att existera.
I våran lösning har vi gjort detta möjligt genom att modifiera brokern man i ett annat chattrum
mqtt://server.se/$subscriptions
kan få information om att chattrum som klienterna ansluter till och kopplar ner sig från. Detta gör att vi nu kan ha ett skript som ligger och väntar på anslutningar och sedan skapar chattrummen dynamisk utifrån klienternas behov.
Med den här funktionen också kan göra ett skript som ligger och väntare på att en klient ska vilja ha ett chattrum som visar någon data som finns på en helt annan server i staden i ett helt annat format. Vårt skript kan nu skapa precis det chattrummet som klienten vill ha genom att kontakta servern i fråga läsa in och formatera om datat och sedan fylla chattrummet med informationen. Vårt skript kan också ligga och övervaka fjärrservern åt oss så länge klienten är ansluten för att uppdatera datat ifall den skulle ändras.
Detta är en funktion som många system idag inte stöder eftersom de bygger på en request-respons modell snarare än en publish subscribe som MQTT gör men som är väldigt bra om man vill ha ett händelsedrivet flöde eller om man tex vill ha nyckeltal som uppdateras i realtid.
Tänk dig ett gata fylld av bibliotek med data:
- I stället för att själv springa runt i varje bibliotek och skaffa nya lånekort och leta i varje hylla och bläddra igenom de böcker du hittar tills du hittar precis den raden du behövde så,
- Går du till datamäklare och ber den skapa ett rum som innehåller precis de sidor rader du behöver.
Så vart kommer datahubben in i bilden?
Säg att staden skapar en broker som likt en kombination av en telefonist och översättare kan koppla dig till alla staden data resurser så som GIS systemet på flera olika avdelningar IoT systemet, ekonomisystemet osv. Trots detta kommer vi troligen vid någon tidpunkt ändå att behöva kontakta flera olika brokers då staden inte har all data som kan vara intressant för ett visst användarfall.
En privat aktör vill ju ha kontroll på sin egen data och kommer på samma sätt som staden att skapa sin egen broker. Vi måste på så sätt förvänta oss att ett dataspace spänner över flera olika brokers.
För att underlätta i mitt program vill jag inte behöva tänka på att jag har med flera olika brokers att göra. Jag vill inte behöva initiera alla olika anslutningar och hantera dessa för att se om anslutningen har gått ner om jag inte längre behöver den osv. Om jag har flera olika objekt i mitt program som behöver data från samma broker vill jag inte heller skapa flera anslutningar till samma broker utan låta de olika objekten dela på den anslutning som redan finns.
Datahubben är en lokal funktion som syftar till att lösa just detta. Låter oss prenumerera på ett chattrum i en broker utan att bry oss om allt det andra. När vi gör prenumerationen så kollar den om det redan finns en anslutning till just den brokern om det inte finns så skapar den en. När vi avprenumererar så behålls anslutningen om andra också använder den men om vi var den enda så avslutas den.
📦 Hur avänder jag Datahubben?
DataHub är ett Python-gränssnitt byggt ovanpå MQTT. Det finns även för Javascript men mer om det i en senare artikel.
Det gör att du kan:
- Subscriba (lyssna) på en dataström från en topic.
- Get (hämta) ett värde en gång.
- Publish (skicka) data till en topic.
Allt detta med hela URL:er i stället för att själv behöva öppna och hålla reda på anslutningar.
Varför är det användbart?
- Du behöver som tidigare nämnts bara en anslutning per server, inte en per widget eller modul.
- Du kan arbeta direkt med fulla adresser som: mqtt://iot.digivis.se/datadirectory/TestArea/signalA eller
mqtt://op-en.se/test/signalB
Utan att bry dig om att de ligger på olika brokers. - Du kan snabbt bygga dashboards, nyckeltal osv. där olika widgets hämtar olika data utan att bry dig om anslutningslogik.
Vad löser den inte?
I nuläget har vi inte någon federerad lösning för åtkomst av de olika brokerna. Som användare måste du därför ladda in dina inloggninguppgifter för vart och ett av de olika brokerna. Det finns flera tänkbara lösningar på detta. En skulle kunna vara en hemma broker som håller koll på dina inloggningar även på de övriga systemen vilket skulle gör att vi bara behöver logga in med ett lösenord och användarnamn eller att man har en central autentiseringsserver alla brokers litar på. Troligtvis kommer vi att behöva en kombination av dessa två då det är orealistiskt att alla organisationer ska lita på varandra. Vi kommer troligen att ha kluster av brokers där de är anslutna till olika autentiseringsservrar men någon enstaka som är helt fristående.
Det finns heller inget sätt att veta hur vi konstruerar adresserna till chattrummen jag behöver veta att jag kan komma åt storleken stadsdelen Kistas yta genom adressen:
mqtt://iot.digivis.se/Namespaces/Providers/dataportalen.stocktolm.se/stadsdelar/Kista/yta
Vi måste skapa konventioner och sökfunktioner för hur den gemensamma adress rymden ska utformas och hur vi navigerar den. Vi kan dock med vårt datanav (Datahub) och datamäklare (brokers) enkelt interagera med de olika resurserna och ta oss till de olika platserna. Vi har så att säga en båt att ta oss dit men inte ännu någon karta.
Illustration: Server → DataHub → Dashboard med flera widgets
📥Installation (Colab/Notebook)
Kör detta först i en ny cell:
pip install dataspace_client
🚀 Komma igång
1. Importera och skapa DataHub
from dataspace_client import datahub
2. Logga in mot en server
datahub.login("mqtt://iot.digivis.se", "username")
Du kommer att frågas efter lösenord. Alternativt kan du använda
datahub.add_credentials("mqtt://iot.digivis.se", "username","password")
Om du tex vill ladda lösenordet från en miljövariabel eller fil.
3. Hämta ett värde (Get)
När vi inte anger en hanterare (handler) till get som kommer den att använda den default hanterare för att tolka svaret som vi tar emot. När det är en katalog listning dvs att topic:et slutar på "/" så kommer den att visualisera kataloger och filer med emojis (se nedan).
Samma gäller för json-data där den gör en pretty print.
datahub.Get("mqtt://iot.digivis.se/datadirectory/TestArea/signalA")
Ger tex. en indexerad utskrift.
Om du vill fånga svaret i en variabel måste du ange en hanterare.
Hanterar funktionen ska ha 3 parametrar topic, payload och en flagga som anger om meddelandet mottogs på den publika eller privata kanalen. Den privata kanalen är en kanal där varje client kan fråga efter historisk data för att komma ikapp. Kommer meddelandet som privat är det alltså ingen ny information för de klienter som redan varit anslutna ett tag. Kommer det via den publika kanaler så är det en ny händelse för hela systemet.
Ett alternativ till att ha en hanterare som bara fungerar på Get är att sätta hanteraren till None. Då kommer Get funktionen att blockera fortsatt körning och vänta tills att vi får svar från servern. När vi får det så returnerar den svaret på vanligt vis. Denna metod gör det lite enklare att skriva kod i vissa fall där man har tid att .
payload = datahub.Get("mqtt://iot.digivis.se/datadirectory/TestArea/signalA",handler = None)
print(payload)
Svaret blir då en binär sträng.
b'{"power":2458,"energy":16298987.923560001,"time":1759326249.557}'
För att använda den om det är JSON vill vi ofta köra
import json
data = json.loads(payload)
print(data["power"])
2458
data variabeln är nu ett objekt som vi kan indexera.
4. Lyssna på ström av data (Subscribe)
def my_handler(topic, payload, private): print("Got data:", topic, payload)datahub.Subscribe("mqtt://iot.digivis.se/datadirectory/TestArea/signalA", my_handler)
5. Skicka data (Publish)
datahub.Publish("mqtt://iot.digivis.se/datadirectory/TestArea/signalA", b"My new value")
🔄 Get vs Subscribe
- Get: Hämtar en gång. Perfekt när du bara vill veta det senaste värdet.
- Subscribe: Håller strömmen öppen och ger dig nya uppdateringar.
Illustration: Get = punkt på tidslinje, Subscribe = kontinuerlig linje.
📊 Data som tabeller
DataHub har inbyggt stöd för Pandas:
df = datahub.GetDataFrame("mqtt://iot.digivis.se/datadirectory/TestArea/logs")print(df.head())
Du kan även hämta loggar vid en viss tidpunkt:
df = datahub.GetDataFrameAt("mqtt://iot.digivis.se/datadirectory/TestArea/logs", ts=1700000000)
🧭 Utforska dataspace
För att se vilka topics som finns:
datahub.Subscribe("mqtt://iot.digivis.se/datadirectory/", handler=None)
Eller lista filer/loggar vid en tidpunkt:
datahub.GetFilesAt("mqtt://iot.digivis.se/datadirectory/TestArea", epoc_time=1700000000)
🔐 Dynamisk säkerhet (DynSec)
DataHub har stöd för dynamisk säkerhet i Mosquitto (DynSec).
Det betyder att du kan skapa användare, roller och grupper direkt via Python.
Exempel: Skapa en ny användare
datahub.add_user_with_role( "mqtt://iot.digivis.se", username="alice", password="mypassword", fullname="Alice Example"
)
Exempel: Skapa en grupp
server = datahub.add_server("mqtt://iot.digivis.se")
dyn = DynSec(server)dyn.create_group("team1")dyn.add_group_client("team1", "alice")dyn.ensure_group_permissions("team1")
⚙️ Så fungerar det under huven
För den som vill förstå detaljerna:
- DataHub håller en lista över servrar (
self.servers). - Varje server representeras av en Broker som ansvarar för anslutning via
paho-mqtt. - När du gör
datahub.Get(url), delas URL:en upp i:- Serveradress (t.ex.
iot.digivis.se) - Topic (t.ex.
datadirectory/TestArea/signalA)
- Serveradress (t.ex.
- Broker prenumererar, tar emot meddelandet och skickar det vidare till din handler eller som returnerat värde.
- Data kan skickas både på publika topics och privata (
$private/...).
🤖 DataHub och AI-integration
Eftersom all data:
- kan hämtas som JSON,
- kan omvandlas till Pandas DataFrames,
- kan visualiseras (t.ex.
.glb-filer med inbyggd 3D-visare),
→ gör det DataHub perfekt för AI-analyser.
En AI kan:
- Hämta loggar med
GetDataFrame, - Träna modeller på historiska data,
- Reagera i realtid på
Subscribe.
📝 Sammanfattning
- DataHub förenklar anslutningar till MQTT/dataspace.
- Du kan Get, Subscribe, Publish med hela URL:er.
- Du slipper göra en anslutning per widget – DataHub hanterar det åt dig.
- Nybörjare kan snabbt komma igång med enkla funktioner.
- Avancerade användare och AI kan nyttja tidsindex, tabeller och dynamisk säkerhet.
👉 Nästa steg:
Testa notebook-exemplet här:
Ny Tutorial för DIGIVIS servern