Hier habe ich ein kleines Script, welches die Senderinformationen von welle-cli zu einem MQTT-Broker der Wahl senden kann.
Dies ermöglicht eine flexible Weiterverarbeitung der Daten in anderen Systemen.
Ich selbst nutze es, um in openHAB die Senderinformationen passend anzeigen zu lassen.
Hierzu werden mehrere Channels eines Items mit den Infos befüllt.
Ich habe gedacht, ich veröffentliche das Script hier mal, falls es jemand gebrauchen und verbessern kann.
Da es sich hierbei um ein wenig „Code-Kram“ handelt, gibts keinerlei Support von mir.
Lest euch das Script durch, die Kommentare sind hoffentlich aussagekräftig genug.
Es sind nur wenige Infos erforderlich, damit das Script geht.
– IP oder Hostname des MQTT-Server
– Port des MQTT-Servers
– Zugangsdaten (Benutzername und Passwort)
– URL der mux.json-Datei von welle-cli
– Phyton3 + pip
# apt install python3-pip # pip3 install paho-mqtt
Speichere dieses Script auf einem Server ab, der die Abfragen durchführen soll:
##################################################################
## Welle-CLI Daten zu MQTT Broker senden
## Erstellt von Hoerli
## 17.12.2024
##################################################################
import json
import requests
import paho.mqtt.client as mqtt
# Konfigurationen
MQTT_BROKER = "192.168.1.111" # IP-Adresse des MQTT-Brokers
MQTT_PORT = 1883 # MQTT Port
MQTT_TOPIC = "welle/mux" # MQTT-Topic
MQTT_USERNAME = "XXXXXXXXXX" # MQTT-Benutzername
MQTT_PASSWORD = "XXXXXXXXXXXXXXX" # MQTT-Passwort
MUX_URL = "http://192.168.1.123:7777/mux.json" # URL der mux.json-Datei
# MQTT-Client einrichten
client = mqtt.Client()
def connect_mqtt():
try:
# Benutzername und Passwort setzen
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.connect(MQTT_BROKER, MQTT_PORT, 60)
print(f"Verbunden mit MQTT-Broker {MQTT_BROKER} auf Port {MQTT_PORT}")
except Exception as e:
print(f"Fehler beim Verbinden mit MQTT-Broker: {e}")
exit(1)
def fetch_mux_data():
try:
# Herunterladen der mux.json-Datei
response = requests.get(MUX_URL)
response.raise_for_status() # Fehler auslösen, wenn die Anfrage fehlschlägt
mux_data = response.json() # JSON-Inhalt parsen
print(f"mux.json erfolgreich von {MUX_URL} abgerufen.")
return mux_data
except requests.exceptions.RequestException as e:
print(f"Fehler beim Abrufen der mux.json: {e}")
exit(1)
except json.JSONDecodeError:
print(f"Fehler beim Parsen der mux.json. Ist es gültiges JSON?")
exit(1)
def publish_mux_data(mux_data):
try:
# CIR Peaks extrahieren
cir_peaks = mux_data.get("cir_peaks", [])
# Demodulator-Daten extrahieren
demodulator = mux_data.get("demodulator", {})
# Ensemble-Informationen extrahieren
ensemble = mux_data.get("ensemble", {})
ensemble_id = ensemble.get("id", "unknown") # Ensemble-ID auslesen oder "unknown" setzen
# Dienste (Services) extrahieren
services = mux_data.get("services", [])
ensemble_services = {}
for service in services:
# Datenrate, Sendername (label), Titel / Show-Infos (dls) und PTY (Genre) senden
service_info = {
"sender_name": service["label"]["label"].strip(),
"show_info": service["dls"]["label"].strip() if service["dls"].get("label") else "",
"bitrate": service["components"][0]["subchannel"]["bitrate"],
"ptystring": service.get("ptystring", "Unknown")
}
# Die 'sid' als Schlüssel für jeden Dienst, um es besser in anderen Anwendungen abfragen zu können
sid = service["sid"]
ensemble_services[sid] = service_info
# Alle Daten kombinieren
combined_payload = {
"cir_peaks": cir_peaks,
"demodulator": demodulator,
"ensemble": ensemble,
ensemble_id: ensemble_services
}
# Daten senden
payload_json = json.dumps(combined_payload, indent=2)
client.publish(MQTT_TOPIC, payload_json)
print(f"Gesendet: {payload_json}")
except Exception as e:
print(f"Fehler beim Verarbeiten der mux.json: {e}")
def main():
connect_mqtt()
mux_data = fetch_mux_data()
publish_mux_data(mux_data)
client.disconnect()
if __name__ == "__main__":
main()
Die Ausgabe ist wie folgt aufgebaut:
– Zuerst gibt es genauere Senderinfos per CIR-Peaks, die nur für Bedarfsfälle übermittelt werden (braucht man eigentlich nicht…)
– Die Demodulations-Infos gebe ich mit, falls ich später mal die Empfangswerte noch irgendwo anzeigen lassen will
– ensemble enthält alle Infos über den gesamten Kanal (Beispiel hier SWR BW S)
– Um in anderen Programmen die Abfragen statisch sauber halten zu können und nichts „gesucht“ werden muss, wird die Kanal-ID als oberster Ast vorangestellt (Beispiel 0x10eb)
– Damit Sender immer über den gleichen Pfad erreichbar sind und nichts „gesucht“ werden muss, wird die Sender-ID als Ast genutzt (Beispiel 0xd3a2)
– Ich habe absichtlich keine Einheiten bei Zahlen mitgegeben, um diese in der Zielanwendung als „number“ statt einem „string“ zu führen
{
"cir_peaks": [
{
"index": 703,
"value": 8.193914413452148
},
],
"demodulator": {
"fic": {
"numcrcerrors": 0
},
"frequencycorrection": 67,
"snr": 14,
"time_last_fct0_frame": 1734445860806
},
"ensemble": {
"ecc": "0xe0",
"id": "0x10eb",
"label": {
"fig2charset": "Undefined",
"fig2label": "",
"fig2rfu": false,
"label": "SWR BW S ",
"shortlabel": "SWR BW S"
}
},
"0x10eb": {
"0xd3a2": {
"sender_name": "SWR Kultur",
"show_info": "",
"bitrate": 128,
"ptystring": "Arts"
},
"0xd301": {
"sender_name": "SWR1 BW",
"show_info": "Bridge to your heart / Wax",
"bitrate": 112,
"ptystring": "Pop Music"
}
}
}
Viel Spaß mit dem Script 🙂
Wusstest du schon, du kannst ganz einfach per openHAB den Kanal von welle-cli setzen lassen.
Einfach das Script nachbauen und mit einem Auslöser ausführen lassen.
welle-cli hat absolut keine Authentifizierung oder ähnliches, daher klappt das wunderbar.