Weather alarm from Royal Netherlands Meteorological Institute (KNMI)

[waltervl]

@FireWizzard,
One hour ago whole of Netherlands was code yellow. The warning for for example Noord-Brabant is already removed now.

I see now the complete coastal area of the Netherlands still have code yellow

[FireWizard]

@waltervl,

Yes, I saw also, that there was a change, but now the province of Groningen shows "Yellow" in the KNMI map and no warning in the Meteoalarm.org map.
This is strange and that makes in my opinion one of them unreliable. I go for KNMI as the best one.

Regards

[Jan Jansen]

Similar to using Nodered, there is also an issue with using DZvents for retrieving weather alerts. I hope for a solution.

Thanks in advance

[FireWizard]

Hello all,

I received an update from KNMI (in Dutch), as follows.

Screenshot_Antwoord_KNMI_Weeralarm.png (36.27 KiB) Viewed 3297 times

As you can see, this does not help us very much.
The URL for "Weerwaarschuwingen" is the one, I already looked at. That one is very complex.

But I discovered, they have now a MQTT server available. I will investigate that one in the next days and keep you updated.

Regards

[EdwinK]

I hope that MQTT will work out. According to the dashboard we still are in code yellow.

[Hansh]

asked the next question in chat gpt

Write a python script to retrieve the weather warning for South Holland from the KNMI data platform api.dataplatform.knmi.nl

the answer was
Of course, here's an example of how you can do that using Python and the `requests` library to retrieve data from the KNMI API:

Code: Select all

python
import requests

def haal_weerwaarschuwing_op():
    # Vul hier je eigen API key in
    api_key = 'jouw_api_sleutel_hier'

    # URL voor de weerwaarschuwing in Zuid-Holland
    url = 'https://api.dataplatform.knmi.nl/open-data/weerwaarschuwingen/v1/warnings?province=Zuid-Holland'

    headers = {'Authorization': f'Bearer {api_key}'}

    # Verzoek om de weerwaarschuwing op te halen
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        weerwaarschuwing = response.json()
        return weerwaarschuwing
    else:
        return None  # Geen gegevens gevonden of fout bij het ophalen

# Oproepen van de functie om de weerwaarschuwing op te halen
weerwaarschuwing_data = haal_weerwaarschuwing_op()

if weerwaarschuwing_data:
    print(weerwaarschuwing_data)
else:
    print("Er zijn geen weerwaarschuwingen gevonden voor Zuid-Holland.")

You must enter your own KNMI API key in the `api_key` variable to access the data. Please note that you must have the correct rights from the KNMI to access their API.

After the holidays I will start working on this to see if anything can be done with it

perhaps a starting point for others

[Hansh]

Still found some time to get started with the code that Chat gpt provided. As is often the case, it is of no use to you, but it does put you on the track.

Found an example on the site https://developer.dataplatform.knmi.nl/open-data-api# that retrieves the last file of the warning dataset. This is a txt file and contains the text of the last warning. Unfortunately there were no warnings at this time so I received a file back with the meaningless text "There are currently no warnings in effect."

The question now is what kind of text is returned with warnings and can it be parsed sufficiently to find local warnings. So wait for bad weather.

I show the code I used. The API key is public and is valid until July 1, 2024. You can create your own API key at: https://developer.dataplatform.knmi.nl/apis

Code: Select all

import logging
import os
import sys

import requests

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(os.environ.get("LOG_LEVEL", logging.INFO))


class OpenDataAPI:
    def __init__(self, api_token: str):
        self.base_url = "https://api.dataplatform.knmi.nl/open-data/v1"
        self.headers = {"Authorization": api_token}

    def __get_data(self, url, params=None):
        return requests.get(url, headers=self.headers, params=params).json()

    def list_files(self, dataset_name: str, dataset_version: str, params: dict):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files",
            params=params,
        )

    def get_file_url(self, dataset_name: str, dataset_version: str, file_name: str):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files/{file_name}/url"
        )


def download_file_from_temporary_download_url(download_url, filename):
    try:
        with requests.get(download_url, stream=True) as r:
            r.raise_for_status()
            with open(filename, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
    except Exception:
        logger.exception("Unable to download file using download URL")
        sys.exit(1)

    logger.info(f"Successfully downloaded dataset file to {filename}")


def main():
    api_key = "eyJvcmciOiI1ZTU1NGUxOTI3NGE5NjAwMDEyYTNlYjEiLCJpZCI6ImE1OGI5NGZmMDY5NDRhZDNhZjFkMDBmNDBmNTQyNjBkIiwiaCI6Im11cm11cjEyOCJ9"
    dataset_name = "waarschuwingen_nederland_48h"
    dataset_version = "1.0"
    logger.info(f"Fetching latest file of {dataset_name} version {dataset_version}")

    api = OpenDataAPI(api_token=api_key)

    # sort the files in descending order and only retrieve the first file
    params = {"maxKeys": 1, "orderBy": "created", "sorting": "desc"}
    response = api.list_files(dataset_name, dataset_version, params)
    if "error" in response:
        logger.error(f"Unable to retrieve list of files: {response['error']}")
        sys.exit(1)

    latest_file = response["files"][0].get("filename")
    logger.info(f"Latest file is: {latest_file}")

    # fetch the download url and download the file
    response = api.get_file_url(dataset_name, dataset_version, latest_file)
    download_file_from_temporary_download_url(response["temporaryDownloadUrl"], latest_file)


if __name__ == "__main__":
    main()

To be continued

[EdwinK]

Neat

I think they should issue a weatheralert now to test

[HvdW]

I wrote a simple dzVents script to handle weerlive data.

Code: Select all

-- Weerlive KNMI gegevens
-- https://weerlive.nl/delen.php , read the 3 tabs
-- API-Key can be obtained at https://weerlive.nl/api/toegang/index.php
-- the json call is like 
-- http://weerlive.nl/api/json-data-10min.php?key=f1a3ab46b8&locatie=52.3456789,6.3456789 -- or just the name of the village you live in

return {
    on = {
        timer = { 'every 10 minutes' },
        httpResponses = { 'WeerDataRetrieved' }
    },
        logging = {
	    level = domoticz.LOG_DEBUG,  -- change to LOG_ERROR to suppress the output in the Domoticz log
	    marker = 'Weerlive DEBUG data'
	},
    execute = function(domoticz, item)
        if (item.isTimer) then
            domoticz.openURL({
            url = 'http://weerlive.nl/api/json-data-10min.php?key=abc123_321cba&locatie=52.3456789,6.3456789',
            method = 'GET',
            callback = 'WeerDataRetrieved'
            })
        elseif (item.isHTTPResponse) then
            if (item.ok) then -- statusCode == 2xx
                -- Display all data from weerlive with the domoticz.utils.dumpTable(item.json) statement
                domoticz.log(domoticz.utils.dumpTable(item.json),domoticz.LOG_DEBUG)
                domoticz.log('De buitentemperatuur is   ' .. item.json.liveweer[1].temp,domoticz.LOG_DEBUG)
                domoticz.log('De gevoelstemperatuur is  ' .. item.json.liveweer[1].gtemp,domoticz.LOG_DEBUG)
                domoticz.log('De luchtvochtigheid is    ' .. item.json.liveweer[1].lv,domoticz.LOG_DEBUG)
                domoticz.log('De weersverwachting is    ' .. item.json.liveweer[1].verw,domoticz.LOG_DEBUG)
                domoticz.log('De weerwaarschuwing is    ' .. item.json.liveweer[1].alarm,domoticz.LOG_DEBUG)
                domoticz.devices('Buitentemperatuur').updateTemperature(item.json.liveweer[1].temp)
                domoticz.devices('Gevoelstemperatuur').updateTemperature(item.json.liveweer[1].gtemp)
                domoticz.devices('Luchtvochtigheid').updateHumidity(item.json.liveweer[1].lv,domoticz.HUM_NORMAL)
                domoticz.devices('Weersverwachting').updateText(item.json.liveweer[1].verw)
                if (item.json.liveweer[1].alarm == '0') then
                    domoticz.devices('Weeralarm').switchOff()
                elseif (item.json.liveweer[1].alarm == '1') then
                    domoticz.devices('Weersverwachting').updateText(item.json.liveweer[1].alarmtxt)
                    domoticz.devices('Weeralarm').switchOn()
                end
            end
        end
    end
}

As you can see I added a switch 'Weeralarm' which will light when there is a weather alarm code.
In case of such a weather alarm the text in 'Weersverwachting' will be replaced by the alarm text instead of the normal weather description text.

Needless to say that this script can be changed in many ways (read the dumpTable output) and it can act as a simple, more adaptable version of the Buienradar plugin.
Of course notifications - which are handled by the Domoticz app on your mobile device - can be inserted in the code as well.

Code: Select all

domoticz.notify('Domoticz notifyer', ' there is a WEATHER ALARM!', domoticz.PRIORITY_HIGH)

The question now is what kind of text is returned with warnings and can it be parsed sufficiently to find local warnings. So wait for bad weather.

I'm curious about the text (item.json.liveweer[1].alarmtxt) as well.
The above code uses your location for collecting data. We'll have to wait for the next code RED, code ORANGE or code YELLOW to see the results.

[Xenomes]

Thanks to all for providing an alternative to the 'Meteo Alarm' plugin that stopped working a while ago.

[Kedi]

A script from late @waaren can be found here viewtopic.php?p=266284#p266284

[Hansh]

Found some time to work on the weather warnings.

In my previous post I showed a script that downloads a text file from the KNMI data platform with the latest weather warning. After some searching I discovered that this text was extracted from the xml file from the same dataset. I can also do what knmi can do, so I adjusted the script from the previous post so that the xml file is also downloaded.

That script has now become:

Code: Select all

import logging
import os
import sys

import requests

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(os.environ.get("LOG_LEVEL", logging.INFO))


class OpenDataAPI:
    def __init__(self, api_token: str):
        self.base_url = "https://api.dataplatform.knmi.nl/open-data/v1"
        self.headers = {"Authorization": api_token}

    def __get_data(self, url, params=None):
        return requests.get(url, headers=self.headers, params=params).json()

    def list_files(self, dataset_name: str, dataset_version: str, params: dict):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files",
            params=params,
        )

    def get_file_url(self, dataset_name: str, dataset_version: str, file_name: str):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files/{file_name}/url"
        )


def download_file_from_temporary_download_url(download_url, filename):
    try:
        with requests.get(download_url, stream=True) as r:
            r.raise_for_status()
            with open(filename, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
    except Exception:
        logger.exception("Unable to download file using download URL")
        sys.exit(1)

    logger.info(f"Successfully downloaded dataset file to {filename}")


def main():
    api_key = "eyJvcmciOiI1ZTU1NGUxOTI3NGE5NjAwMDEyYTNlYjEiLCJpZCI6ImE1OGI5NGZmMDY5NDRhZDNhZjFkMDBmNDBmNTQyNjBkIiwiaCI6Im11cm11cjEyOCJ9"
    dataset_name = "waarschuwingen_nederland_48h"
    dataset_version = "1.0"
    logger.info(f"Fetching latest file of {dataset_name} version {dataset_version}")

    api = OpenDataAPI(api_token=api_key)

    # sort the files in descending order and only retrieve the first file
    #--- maxkeys aangepast naar 2 ipv 1
    params = {"maxKeys": 2, "orderBy": "created", "sorting": "desc"}
    response = api.list_files(dataset_name, dataset_version, params)
    if "error" in response:
        logger.error(f"Unable to retrieve list of files: {response['error']}")
        sys.exit(1)

    # last txt file
    latest_file = response["files"][0].get("filename")
    logger.info(f"Latest file is: {latest_file}")

    #last xml file
    latest_file_xml = response["files"][1].get("filename")
    logger.info(f"Latest file_xml is: {latest_file_xml}")

    # fetch the download url and download the file
    response = api.get_file_url(dataset_name, dataset_version, latest_file)
    download_file_from_temporary_download_url(response["temporaryDownloadUrl"], latest_file)

    # fetch the download url and download the file_xml
    response = api.get_file_url(dataset_name, dataset_version, latest_file_xml)
    download_file_from_temporary_download_url(response["temporaryDownloadUrl"], latest_file_xml)


if __name__ == "__main__":
    main()

After I downloaded the XML file, I started working on it. XML is not my expertise (actually I don't know much about it). I do know that it is a tree structure. When examining the XML it seemed to have an unclear structure and the desired data was difficult to find. After a tip from my son to load the file in Excel, I did this. The tree structure became a lot clearer. It contained all the necessary data from the past 48 hours. My findings at a glance:
It starts with metadata - we don't need this right away
The desired data can be found for every hour up to 48 hours in the following element:
"root/data/cube/timeslice"
So an element for every hour. I am concerned about the last hour and this can be retrieved with an xpath instruction:
"root.find("./data/cube/timeslice")"
Here's where it gets confusing. The data is not grouped by location but by phenomenon (wind, rain, etc.). After some fiddling around, I arrived at the following xpath instruction, which retrieves all information for the province of South Holland (ZH).
"data.findall("./phenomenon/location/[location_id='ZH']")"

The script that extracts the data looks like this:

Code: Select all

import xml.etree.ElementTree as ET

def main():
    doc= ET.parse("knmi_waarschuwingen.xml")
    root =doc.getroot()
    data = root.find("./data/cube/timeslice")
    data = data.findall("./phenomenon/location/[location_id='ZH']")
    print("DATA")
   
    for elem in data:
        print (f"->{elem.tag}")              
        for subelem in elem:
            print (f"-->>{subelem.tag} : {subelem.text}")
            for sub2elem in subelem:
                print (f"--->>> {sub2elem.tag} : {sub2elem.text}")


if __name__ == "__main__":
    main()
    print("end")

The script assumes that there is an XML document with the name "knmi_waarschuwingen.xml". this can be retrieved with the modified script from my previous post.

The data has now been retrieved. the next challenge is to get this to domoticz.
and if successful, bring the scripts together and adjust them so that you can choose a location.

tips, tricks and positive feedback are very welcome

to be continued

[Hansh]

Yeh i make it run

After a lot of fiddling around how to read an XML file (I had no knowledge of it at all), I managed to read and parse the data from the weather warning file.
The problem was to only extract the data that was relevant for a province and not the data for the entire Netherlands.
The structure used by the KNMI was not arranged by warning level but by time slot -> weather phenomenon -> province and then by warning level.
This is exactly the opposite of how I wanted it.

In the code below we managed to extract the warning level with the associated text for a province and transfer it to domoticz

For the code to work it is necessary that a correct API_KEY is specified. The API_KEY that is currently included is a public key and is valid until June 2024.
You can request a personal key or obtain a new public key on the KNMI website
https://developer.dataplatform.knmi.nl/open-data-api

You must enter your own value for the variables below
URL_domoticz = "http://xx.xx.xx.xx:yyyy"
device_id="xx"
province= "ZH" (here your own province choice between: "WAE, GR, FR, DR, NH, FL, OV, GL, UT, ZH, ZE, NB, LB, WAB")

You can run the script periodically via a cron job. e.g. every hour

The code is far from optimal, so if there are forum members who see improvements in it, I would like to see them.
Part of the code is inspired by examples from the KNMI, in particular retrieving the files

In Domoticz I used the device general alert

Code: Select all

#! /usr/bin/env python3

# A script to display the weather warning issued by the KNMI in Domotcz.
# Written By J. Hotting

# The data is available on the KNMI datpltform "https://dataplatform.knmi.nl/"
# To make the program work you need a api_key, which can be obtained from the data platform "https://developer.dataplatform.knmi.nl/open-data-api"
# The key used is a public api_key and is valid until June 2024
# You must specify which area/province you want to see notifications from
# You can choose between: WAE, GR, FR, DR, NH, FL, OV, GL, UT, ZH, ZE, NB, LB, WAB

import os
import logging
import sys
import xml.etree.ElementTree as ET
import requests
from time import sleep

logging.basicConfig(format='%(asctime)s: %(levelname)s: %(message)s',datefmt='%Y-%m-%d %H:%M:%S')
logger = logging.getLogger(__name__)
logger.setLevel(os.environ.get("LOG_LEVEL", logging.ERROR))



api_key = "eyJvcmciOiI1ZTU1NGUxOTI3NGE5NjAwMDEyYTNlYjEiLCJpZCI6ImE1OGI5NGZmMDY5NDRhZDNhZjFkMDBmNDBmNTQyNjBkIiwiaCI6Im11cm11cjEyOCJ9"
dataset_name = "waarschuwingen_nederland_48h"
dataset_version = "1.0"
URL_domoticz = "http://192.168.0.5:8081"
device_id="208"
province= "ZH"

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

class OpenDataAPI:
    def __init__(self, api_token: str):
        self.base_url = "https://api.dataplatform.knmi.nl/open-data/v1"
        self.headers = {"Authorization": api_token}

    def __get_data(self, url, params=None):
        return requests.get(url, headers=self.headers, params=params).json()

    def list_files(self, dataset_name: str, dataset_version: str, params: dict):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files",
            params=params,
        )

    def get_file_url(self, dataset_name: str, dataset_version: str, file_name: str):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files/{file_name}/url"
        )


def download_file_from_temporary_download_url(download_url, filename):
    try:
        with requests.get(download_url, stream=True) as r:
            r.raise_for_status()
            with open(filename, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
    except Exception:
        logger.exception("Unable to download file using download URL")
        sys.exit(1)

    #logger.info(f"Successfully downloaded dataset file to {filename}")


def Get_knmi_files():

    logger.info(f"Fetching latest file of {dataset_name} version {dataset_version}")

    api = OpenDataAPI(api_token=api_key)

    # sort the files in descending order and only retrieve the first file
    #--- maxkeys aangepast naar 2 ipv 1
    params = {"maxKeys": 2, "orderBy": "created", "sorting": "desc"}
    FileleList = api.list_files(dataset_name, dataset_version, params)

    if "error" in FileleList:
        logger.error(f"Unable to retrieve list of files: {response['error']}")
        sys.exit(1)

    for x in range(2):
        #get file name       
        latest_file = FileleList["files"][x].get("filename")
        logger.info(f"Latest {latest_file[-4:]} file is: {latest_file}")
        # fetch the download url and download the file
        response = api.get_file_url(dataset_name, dataset_version, latest_file)
        TempFilename= latest_file[:19]+latest_file[-4:]
        download_file_from_temporary_download_url(response["temporaryDownloadUrl"], TempFilename)

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def transfer_to_Domoticz(ID,status,text):
    #execute request and handle errors
    #convert the status to the domoticz requirements (add 1 to the status)
    status = str(int(status)+1)
    # send the data to domoticz
    url= f"{URL_domoticz}/json.htm?type=command&param=udevice&idx={ID}&nvalue={status}&svalue={str(text)}"
    try:
        response=requests.get(url)
    except:
        logger.exception("An error occurred while writing to domoticz")
        sys.exit(1)
    


def parse_XML_file(File):
    #get data from file
    doc= ET.parse(File)
    root = doc.getroot()
    # filter the data for the desired area
    data = root.findall(f"./data/cube/timeslice/phenomenon/location/[location_id='{province}']/")
    warningStatus = 0
    WarningList = []
    WarningText =""
    status=False
    # loop through the selected data and process the outcome
    for loc in data:
        # if the warning level is higher than 0
        if loc.tag == "location_warning_status" and int(loc.text) > 0 :
            status=True # flag to mark whether text needs to be processed
            
            # determine the highest warning level
            if warningStatus < int(loc.text): 
                warningStatus= int(loc.text)
        if status and loc.tag == "text": # process text
            status=False
            for txt in loc:
                if txt.tag == "text_data" and txt.text != None : WarningList.append(txt.text) # put the text in the list

    # remove duplicates from the WarningList list
    WarningList = list(dict.fromkeys(WarningList)) 

    #transform list to a string
    if WarningList: WarningText = "\n".join(WarningList)

    return warningStatus,WarningText

if __name__ == "__main__":
    Get_knmi_files()
    x,y =parse_XML_file("knmi_waarschuwingen.xml")
    transfer_to_Domoticz (device_id,x,y)
    logger.info("The program has ended successfully")

[Vomera]

Yeh i make it run

After a lot of fiddling around how to read an XML file (I had no knowledge of it at all), I managed to read and parse the data from the weather warning file.
The problem was to only extract the data that was relevant for a province and not the data for the entire Netherlands.
The structure used by the KNMI was not arranged by warning level but by time slot -> weather phenomenon -> province and then by warning level.
This is exactly the opposite of how I wanted it.

In the code below we managed to extract the warning level with the associated text for a province and transfer it to domoticz

For the code to work it is necessary that a correct API_KEY is specified. The API_KEY that is currently included is a public key and is valid until June 2024.
You can request a personal key or obtain a new public key on the KNMI website
https://developer.dataplatform.knmi.nl/open-data-api

You must enter your own value for the variables below
URL_domoticz = "http://xx.xx.xx.xx:yyyy"
device_id="xx"
province= "ZH" (here your own province choice between: "WAE, GR, FR, DR, NH, FL, OV, GL, UT, ZH, ZE, NB, LB, WAB")

You can run the script periodically via a cron job. e.g. every hour

The code is far from optimal, so if there are forum members who see improvements in it, I would like to see them.
Part of the code is inspired by examples from the KNMI, in particular retrieving the files

In Domoticz I used the device general alert

Code: Select all

#! /usr/bin/env python3

# A script to display the weather warning issued by the KNMI in Domotcz.
# Written By J. Hotting

# The data is available on the KNMI datpltform "https://dataplatform.knmi.nl/"
# To make the program work you need a api_key, which can be obtained from the data platform "https://developer.dataplatform.knmi.nl/open-data-api"
# The key used is a public api_key and is valid until June 2024
# You must specify which area/province you want to see notifications from
# You can choose between: WAE, GR, FR, DR, NH, FL, OV, GL, UT, ZH, ZE, NB, LB, WAB

import os
import logging
import sys
import xml.etree.ElementTree as ET
import requests
from time import sleep

logging.basicConfig(format='%(asctime)s: %(levelname)s: %(message)s',datefmt='%Y-%m-%d %H:%M:%S')
logger = logging.getLogger(__name__)
logger.setLevel(os.environ.get("LOG_LEVEL", logging.ERROR))



api_key = "eyJvcmciOiI1ZTU1NGUxOTI3NGE5NjAwMDEyYTNlYjEiLCJpZCI6ImE1OGI5NGZmMDY5NDRhZDNhZjFkMDBmNDBmNTQyNjBkIiwiaCI6Im11cm11cjEyOCJ9"
dataset_name = "waarschuwingen_nederland_48h"
dataset_version = "1.0"
URL_domoticz = "http://192.168.0.5:8081"
device_id="208"
province= "ZH"

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

class OpenDataAPI:
    def __init__(self, api_token: str):
        self.base_url = "https://api.dataplatform.knmi.nl/open-data/v1"
        self.headers = {"Authorization": api_token}

    def __get_data(self, url, params=None):
        return requests.get(url, headers=self.headers, params=params).json()

    def list_files(self, dataset_name: str, dataset_version: str, params: dict):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files",
            params=params,
        )

    def get_file_url(self, dataset_name: str, dataset_version: str, file_name: str):
        return self.__get_data(
            f"{self.base_url}/datasets/{dataset_name}/versions/{dataset_version}/files/{file_name}/url"
        )


def download_file_from_temporary_download_url(download_url, filename):
    try:
        with requests.get(download_url, stream=True) as r:
            r.raise_for_status()
            with open(filename, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
    except Exception:
        logger.exception("Unable to download file using download URL")
        sys.exit(1)

    #logger.info(f"Successfully downloaded dataset file to {filename}")


def Get_knmi_files():

    logger.info(f"Fetching latest file of {dataset_name} version {dataset_version}")

    api = OpenDataAPI(api_token=api_key)

    # sort the files in descending order and only retrieve the first file
    #--- maxkeys aangepast naar 2 ipv 1
    params = {"maxKeys": 2, "orderBy": "created", "sorting": "desc"}
    FileleList = api.list_files(dataset_name, dataset_version, params)

    if "error" in FileleList:
        logger.error(f"Unable to retrieve list of files: {response['error']}")
        sys.exit(1)

    for x in range(2):
        #get file name       
        latest_file = FileleList["files"][x].get("filename")
        logger.info(f"Latest {latest_file[-4:]} file is: {latest_file}")
        # fetch the download url and download the file
        response = api.get_file_url(dataset_name, dataset_version, latest_file)
        TempFilename= latest_file[:19]+latest_file[-4:]
        download_file_from_temporary_download_url(response["temporaryDownloadUrl"], TempFilename)

#++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

def transfer_to_Domoticz(ID,status,text):
    #execute request and handle errors
    #convert the status to the domoticz requirements (add 1 to the status)
    status = str(int(status)+1)
    # send the data to domoticz
    url= f"{URL_domoticz}/json.htm?type=command&param=udevice&idx={ID}&nvalue={status}&svalue={str(text)}"
    try:
        response=requests.get(url)
    except:
        logger.exception("An error occurred while writing to domoticz")
        sys.exit(1)
    


def parse_XML_file(File):
    #get data from file
    doc= ET.parse(File)
    root = doc.getroot()
    # filter the data for the desired area
    data = root.findall(f"./data/cube/timeslice/phenomenon/location/[location_id='{province}']/")
    warningStatus = 0
    WarningList = []
    WarningText =""
    status=False
    # loop through the selected data and process the outcome
    for loc in data:
        # if the warning level is higher than 0
        if loc.tag == "location_warning_status" and int(loc.text) > 0 :
            status=True # flag to mark whether text needs to be processed
            
            # determine the highest warning level
            if warningStatus < int(loc.text): 
                warningStatus= int(loc.text)
        if status and loc.tag == "text": # process text
            status=False
            for txt in loc:
                if txt.tag == "text_data" and txt.text != None : WarningList.append(txt.text) # put the text in the list

    # remove duplicates from the WarningList list
    WarningList = list(dict.fromkeys(WarningList)) 

    #transform list to a string
    if WarningList: WarningText = "\n".join(WarningList)

    return warningStatus,WarningText

if __name__ == "__main__":
    Get_knmi_files()
    x,y =parse_XML_file("knmi_waarschuwingen.xml")
    transfer_to_Domoticz (device_id,x,y)
    logger.info("The program has ended successfully")

Hi Thank you for finding out the nice code! Can you show me your domoticz logs, i just install it, but get no text at the moment. Maybe because there is no warning at the moment, but the text file knmi_waarschuwingen.txt shows :

Code: Select all

Er zijn geen waarschuwingen

Er zijn momenteel geen waarschuwingen van kracht.

Opgesteld door het KNMI op: maandag 22 januari 2024 14:51 uur

[Hansh]

It may be true that you do not see any text.

First, I see from your screenshot that you used a text device to pass on the warnings. The script assumes that a "general Alert" device has been created. You will then also see a colored triangle, green, yellow, orange or red respectively.

Schermafbeelding 2024-01-22 192808.png (15.74 KiB) Viewed 1885 times

In domoticz I don't get any log messages. In the log file I see the following lines

2024-01-22 19:27:37: INFO: Fetching latest file of warnings_nederland_48h version 1.0
2024-01-22 19:27:38: INFO: Latest .txt file is: knmi_alerts_202401221636.txt
2024-01-22 19:27:40: INFO: Latest .xml file is: knmi_Warnings_202401221635.xml

The text of warnings.txt is a text compiled by the knmi for the entire country and I do not use it. But in their data platform the order of the files is by date and it may be that the xml file is second on the list. By downloading two I always have the last xml file. The txt file is by-catch

To improve the script, I delved further into XML files, so I can improve the script. In the version I am currently working on, the times during which the warning applies are also displayed. Furthermore, the province can be selected using a domoticz variable and logging will go to the correct place on the raspberry nl /var/log/scripts.

When the new script is completely to my liking, I will post it.

The knmi also provides MQTT messages when a file has been modified. The goal is to include this in the script so that files are only retrieved if new ones are available. This still has some snags, but it will work

If you want to try/test the rough (working) version, send me a PM and I can send it to you (this is possible in Dutch)

Greetings Hans

[Vomera]

I guess its working now because see this in the logging. I wil change it back to general alert (use it with the previous script).

[Hansh]

I have continued to improve my script from my Jan 4, 2024 post.
A number of changes and improvements made include:
- the time period of the warning is now also visible
- the script runs in an infinite loop and does not need to be started periodically
- A KNMI file is only downloaded if a new file is available
- Receives a message via MQTT from the KNMI that a new file is available

All this has a price. Due to the security at the KNMI, multiple keys are required, which you can obtain from the KNMI.
Domoticz's security also requires a key

EXPLANATION

In the script, a number of variables are defined in the initialize global variables section.
Some of these require a personal value. The script explains how this value can be obtained.
These variables have been given the value "Create your own" in the script. change this to your own value
The URL for Domoticz must also be specified in the variable "URL_DOMOTICZ"

The log file is placed in the same folder in which the script is running.
If you want to put it in the default folder for log files, change OUTPUTFILE to: OUTPUTFILE = f"/var/log/{SCRPT_NAME}.log"

To start the script on every reboot, add the following line to the crontab file with the command: crontab -e
@reboot sleep 120 && /usr/bin/python3 -u PATH TO SCRIPT
(example path to script :/home/pi/knmi_alert/WeatherAlertToDomoticz.py)

The MQTT modules are not always installed in Python3, this can be done manually with "pip install paho-mqtt"

In domoticz a device of the alert type is expected

alert_device.jpg (14 KiB) Viewed 1091 times

After the program has started, a message appears stating that no data has been found. This message will remain until a new file is issued by the KNMI and can take up to 5 hours

Code: Select all

import os
import sys
import ssl
import json
import logging
import requests
from time import sleep
from queue import Queue
from datetime import datetime
import xml.etree.ElementTree as ET
import paho.mqtt.client as mqtt_client
import paho.mqtt.properties as properties
from logging.handlers import RotatingFileHandler

  
# =============================================================================
# ===================== initialize global variabelen ==========================
# ============================================================================= 

# global variable in use for mqtt
BROKER_DOMAIN = "mqtt.dataplatform.knmi.nl"
# Client ID should be made static, it is used to identify your session, so that
# missed events can be replayed after a disconnect
# https://www.uuidgenerator.net/version4
CLIENT_ID = "Create your own"
# Obtain your token at: https://developer.dataplatform.knmi.nl/notification-service
TOKEN = "Create your own"
# This will listen to both file creation and update events of this dataset:
# https://dataplatform.knmi.nl/dataset/waarschuwingen-nederland-48h-1-0
# This topic has at least 1 event every 5 hours. Usually more
TOPIC = "dataplatform/file/v1/waarschuwingen_nederland_48h/1.0/#"
# Version 3.1.1 also supported
PROTOCOL = mqtt_client.MQTTv5

# global variable in use when retrieving xml file
# Obtain your token at: https://developer.dataplatform.knmi.nl/open-data-api#token
API_KEY = "Create your own"
DATASET_NAME = "waarschuwingen_nederland_48h"
DATASET_VERSION = "1.0"

# variables related to domoticz
# The authorization requirements can be found at
# https://www.domoticz.com/wiki/Domoticz_API/JSON_URL%27s#Authorization
# user: domoticz pasw: class AUTORISATIE_TOKEN => "Basic ZG9tb3RpY3o6Y2xhc3M="
AUTORISATIE_TOKEN = "Create your own" 
DEVICE_ID="Create your own" # number of the device in domoticz
PROVINCE= "ZH" # default value. You can choose between: WAE, GR, FR, DR, NH, FL, OV, GL, UT, ZH, ZE, NB, LB, WAB
WEEKDAYS=["ma","di","wo","do","vr","za","zo"] # list to determine weekdays
URL_DOMOTICZ = "http://xxx.xxx.xxx.xxx:8080"

# other global variables    
# create a queue for the available files 
File_Q = Queue()

# script name. Used for log file and temporary file names
SCRPT_NAME = os.path.basename(sys.argv[0])[:-3]
OUTPUTFILE = f"{SCRPT_NAME}.log"
    
# waiting time between script executions
WAIT_EXECUTION = 300 # 5 MIN
# number of wait cycles before domoticz is refreshed (3600/wait time)
WAIT_CYCLES =  int(3600/WAIT_EXECUTION)-1

# =============================================================================  
# *****************************************************************************
# ****************** don't change anything below this line ********************  
# *****************************************************************************
# =============================================================================
  
  
  
    
# =============================================================================
# ========================== initialize logging ===============================
# =============================================================================  

logging.basicConfig(
    format='%(asctime)s: %(levelname)-8s: %(funcName)-12s:  %(lineno)-4d: %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    level = logging.INFO,
    )
logger = logging.getLogger(__name__)

# Define a handler witch writes INFO messages or higher to a file and rotate at max bytes
fh = RotatingFileHandler(OUTPUTFILE,maxBytes=102400, backupCount=2) # max 0.1 mb
fh.setLevel(logging.INFO)
# set a format which is for File use
ff = logging.Formatter('%(asctime)s: %(levelname)-8s: %(funcName)-12s: %(message)s','%Y-%m-%d %H:%M:%S',)
fh.setFormatter(ff)
# add the handler to the root logger
logger.addHandler(fh)

logger.info ("="*52)
logger.info ("="*18 + " Script started " + "="*18)
logger.info ("="*52)

# ============================================================================= 
# ================================ Classes ==================================== 
# =============================================================================
  
class Domoticz:
    """
    Class that simplifies communication with domoticz.
    The authorization requirements can be found at
    https://www.domoticz.com/wiki/Domoticz_API/JSON_URL%27s#Authorization

    __get_data provides the communication with domoticz. This function is called by
    the other functions that configure the appropriate parameters
    
    Functions:
    Status => requests the status of a device with a specified ID
    update => update the value of the specified device with the specified values
                which value is expected depends on the device and can be found at:
                https://www.domoticz.com/wiki/Domoticz_API/JSON_URL%27s#Update_devices.2Fsensors
    log     => put a message in the domoticz log
    variable=> requests the value of a domoticz variable with the specified ID
    """
    
    def __init__(self, url:str = None , id:int = None) -> None:
        self.base_url = f"{URL_DOMOTICZ}/json.htm" if not url else url
        self.headers = {"Authorization":AUTORISATIE_TOKEN}
        self.id=id
        self.type='command'
    
    def __get_data(self, url:str, params:dict=None):
        try:
            with requests.get(url, headers=self.headers, params=params) as r:
                logger.info(f"=> Domoticz: {r.url}")            
                return r.json()
        except Exception as err:
            logger.error()(f"Error handeling domoticz \n{err}")
            return"ERROR"
    
    def status(self,id:int=None):
        id = self.id if not id else id
        return self.__get_data(self.base_url,dict(
           type =self.type,
           param ='getdevices',
           rid = id
        ))
        
    def update(self,id:int= None, nvalue:float=0, svalue:str=""):
        id = self.id if not id else id        
        return self.__get_data(self.base_url,dict(
           type =self.type,
           param ='udevice',
           idx = id,
           nvalue = nvalue,
           svalue = svalue
        ))
        
    def log (self, mess, level:int = 2):
        return (self.__get_data(self.base_url,dict(
           type =self.type,
           param ='addlogmessage',
           message = mess,
           level = level
        )))
    
    def variabele (self, id:int=1):        
        return self.__get_data(self.base_url,dict(
           type =self.type,
           param ='getuservariable',
           idx = id
        ))
        
class OpenDataAPI:
    def __init__(self, api_token: str):
        self.base_url = "https://api.dataplatform.knmi.nl/open-data/v1"
        self.headers = {"Authorization": api_token}

    def __get_data(self, url, params=None):
        return requests.get(url, headers=self.headers, params=params).json()

    def list_files(self, DATASET_NAME: str, DATASET_VERSION: str, params: dict):
        return self.__get_data(
            f"{self.base_url}/datasets/{DATASET_NAME}/versions/{DATASET_VERSION}/files",
            params=params,
        )

    def get_file_url(self, temp_url: str):
        return self.__get_data(
            f"{temp_url}"
        )

# =============================================================================
# ============================== Functions ====================================
# =============================================================================

def connect_mqtt() -> mqtt_client:
    def on_connect(c: mqtt_client, userdata, flags, rc, reason_code, props=None):
        logger.info(f"Connected using client ID: {str(c._client_id)}")
        logger.info(f"Session present: {str(flags['session present'])}")
        logger.info(f"Connection result: {str(rc)}")
        # Subscribe here so it is automatically done after disconnect
        subscribe(c, TOPIC)

    client = mqtt_client.Client(client_id=CLIENT_ID, protocol=PROTOCOL, transport="websockets")
    client.tls_set(tls_version=ssl.PROTOCOL_TLS)
    connect_properties = properties.Properties(properties.PacketTypes.CONNECT)
    # Maximum is 3600
    connect_properties.SessionExpiryInterval = 3600

    # The MQTT username is not used for authentication, only the token
    username = "token"
    client.username_pw_set(username, TOKEN)
    client.on_connect = on_connect

    client.connect(host=BROKER_DOMAIN, port=443, keepalive=60, clean_start=False, properties=connect_properties) # set the clean_start to false or true. to delete previous sessions

    return client

def subscribe(client: mqtt_client, topic: str):
    def on_message(c: mqtt_client,userdata , message):
        # NOTE: Do NOT do slow processing in this function, as this will interfere with PUBACK messages for QoS=1.
        # A couple of seconds seems fine, a minute is definitely too long.
        #logger.info(f"Received message on topic {message.topic}: {str(message.payload)}")
        load = json.loads(message.payload)        
        if 'data'in load:
            if 'xml' in load['data']['filename']:
                # Place the available file in a queue
                File_Q.put(load['data']['url'])
        else:
            logger.info(f"Received message on topic: {message.topic}: {str(message.payload)}")
            
    def on_subscribe(c: mqtt_client, userdata, mid, granted_qos, *other):
        logger.info(f"Subscribed to topic '{topic}'")

    client.on_subscribe = on_subscribe
    client.on_message = on_message
    # A qos=1 will replay missed events when reconnecting with the same client ID. Use qos=0 to disable
    client.subscribe(topic, qos=1)


def get_knmi_files():
    logger.info(f"Fetching latest file of {DATASET_NAME} version {DATASET_VERSION}")
    #download_url='https://api.dataplatform.knmi.nl/open-data/v1/datasets/waarschuwingen_nederland_48h/versions/1.0/files/knmi_waarschuwingen_202402120747.xml/url'
    download_url= File_Q.get()
    
    api = OpenDataAPI(api_token=API_KEY)
    
    try:
        response = api.get_file_url(download_url)
        with requests.get(response['temporaryDownloadUrl'], stream=True) as r:
            r.raise_for_status()
            with open(f"{SCRPT_NAME}.xml", "wb") as f:
                for chunk in r.iter_content(chunk_size=8192): # kan naar 1024 voor minder mem gebruik
                    f.write(chunk)
    except Exception:
        logger.error("Unable to download file using download URL")
        sys.exit('system error. Program terminated')
 
def parse_XML_file(File):
    # Check if data is available
    if  not os.path.isfile(File): return (0, "No data available \nThis can take up to 5 hours")
    #get data from file
    doc= ET.parse(File)
    root = doc.getroot()
    #set local variabelen
    warningStatus = 0
    WarningList = []
    WarningText =""
    status=False
    startTime=""
    endTime=""

    # loop through the selected data and process the outcome    
    for timeslot in root.findall(".//data/cube/timeslice//"):
        # save the time of the timeslot being processed
        if timeslot.tag == "timeslice_id":
            TimeslotTime = timeslot.text
        # process all phenomena in the timeslot
        if timeslot.tag == "phenomenon":
            for phenomenon in timeslot:
                # process all locations in phenomena
                if phenomenon.tag == "location":                     
                    for locations in phenomenon:
                        # proces wanted location in locations
                        if locations.tag == "location_id" and locations.text != PROVINCE: break # if not Wanted area then exit for loop
                        if locations.tag =="location_warning_status" and int(locations.text) > 0: # if a warning is active
                            status= True # flag to mark whether text needs to be processed

                            # sets the highest warning level in warningStatus
                            if warningStatus < int(locations.text): warningStatus= int(locations.text) 

                            # proces the warning text
                        if status and locations.tag == "text": # process text
                            status = False 
                            logger.debug(f"Time slot with warning: {TimeslotTime}")
                            # manage start and end time
                            if startTime == "": startTime= datetime.fromisoformat(TimeslotTime)
                            endTime = datetime.fromisoformat(TimeslotTime)
                            # Get the desired text
                            for txt in locations:
                                if txt.tag == "text_data" and txt.text != None : WarningList.append(txt.text) # put the text in the list
    
    
    # loop through the selected data and process the outcome
    if startTime : # if a time is available, include it in the WarningText                               
        WarningText= f"Van {WEEKDAYS[datetime.weekday(startTime)]} {startTime.hour}:00 tot {WEEKDAYS[datetime.weekday(endTime)]} {endTime.hour + 1}:00 \n"

    # remove duplicates from the WarningList list
    WarningList = list(dict.fromkeys(WarningList)) 

    #transform list to a string
    if WarningList: WarningText += "\n".join(WarningList)

    if not WarningText: WarningText ="Geen waarschuwingen"
   
    return warningStatus,WarningText   


def run():
    wait_cycles = 1
    client = connect_mqtt()
    client.loop_start()
    wwd=Domoticz(id=DEVICE_ID)
    Wstatus,Wtext = parse_XML_file(f"{SCRPT_NAME}.xml")
    wwd.update(nvalue=Wstatus,svalue=Wtext) # send to domoticz
    while True:
        while not File_Q.empty(): # execute when new files are available
            get_knmi_files()
            wait_cycles = WAIT_CYCLES+1
        if wait_cycles >= WAIT_CYCLES:  # run once an hour or when there are new files         
            Wstatus,Wtext = parse_XML_file(f"{SCRPT_NAME}.xml")
            #convert the status to the domoticz requirements (add 1 to the status)
            Wstatus += 1
            wwd.update(nvalue=Wstatus,svalue=Wtext) # send to domoticz
            wwd.log(f'{SCRPT_NAME}: Has updated device {DEVICE_ID}')
            wait_cycles = 0
        else:
            wait_cycles += 1
        sleep(WAIT_EXECUTION) # time for a cup of coffee.....or a nap

if __name__ == "__main__":
    run()

If you have any comments or wishes, please let me know.
Kind regards, HansH

[EdwinK]

I did something wrong (as usually) because this is the output I get

Code: Select all

 python3 WeatherAlert.py
2024-03-07 11:31:31: INFO    : <module>    :  92  : ====================================================
2024-03-07 11:31:31: INFO    : <module>    :  93  : ================== Script started ==================
2024-03-07 11:31:31: INFO    : <module>    :  94  : ====================================================
Exception ignored in: <function Client.__del__ at 0x75d19df0>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 874, in __del__
    self._reset_sockets()
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 1133, in _reset_sockets
    self._sock_close()
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 1119, in _sock_close
    if not self._sock:
AttributeError: 'Client' object has no attribute '_sock'
Traceback (most recent call last):
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 335, in <module>
    run()
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 314, in run
    client = connect_mqtt()
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 197, in connect_mqtt
    client = mqtt_client.Client(client_id=CLIENT_ID, protocol=PROTOCOL, transport="websockets")
TypeError: __init__() missing 1 required positional argument: 'callback_api_version'

[Hansh]

I did something wrong (as usually) because this is the output I get

Code: Select all

 python3 WeatherAlert.py
2024-03-07 11:31:31: INFO    : <module>    :  92  : ====================================================
2024-03-07 11:31:31: INFO    : <module>    :  93  : ================== Script started ==================
2024-03-07 11:31:31: INFO    : <module>    :  94  : ====================================================
Exception ignored in: <function Client.__del__ at 0x75d19df0>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 874, in __del__
    self._reset_sockets()
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 1133, in _reset_sockets
    self._sock_close()
  File "/usr/local/lib/python3.9/dist-packages/paho/mqtt/client.py", line 1119, in _sock_close
    if not self._sock:
AttributeError: 'Client' object has no attribute '_sock'
Traceback (most recent call last):
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 335, in <module>
    run()
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 314, in run
    client = connect_mqtt()
  File "/home/pi/WeatherAlert/WeatherAlert.py", line 197, in connect_mqtt
    client = mqtt_client.Client(client_id=CLIENT_ID, protocol=PROTOCOL, transport="websockets")
TypeError: __init__() missing 1 required positional argument: 'callback_api_version'

It looks like the problem is in the mqtt client module. The version I installed and everything works with is
"Requirement already satisfied: paho-mqtt in ./.local/lib/python3.9/site-packages (1.6.1)"

The problem is also described here
https://stackoverflow.com/questions/779 ... sion-error