Aviso notification for DT data availability

References:

Credit:

The pyaviso package is provided by the European Centre for Medium-Range Weather Forecasts (ECMWF).

This notebook shows how to check the data availablility for the Weather-Induced Extremes Digital Twin (Extremes DT) using the ECMWF Aviso package.

Install the pyaviso package¶

pip  install  pyaviso --upgrade --quiet

Note: you may need to restart the kernel to use updated packages.

Import pyaviso¶

Import pyaviso and other useful libraries. Defining constants and functions.

from datetime import datetime
from pprint import pprint as pp

from pyaviso import NotificationManager, user_config

LISTENER_EVENT = "data"  # Event for the listener, options are mars and dissemination
TRIGGER_TYPE = "function"  # Type of trigger for the listener

Defining the data to be notified¶

The following request describes the data whose availability we want to be notified

REQUEST = {
    "class": "d1",
    "expver": "0001",
    "stream": "wave",
    "type": "fc",
    "time": "00",
    "step": "0",
    "levtype": "sfc",
#    "levelist": "",
#    "param": "168"   
}  # Request configuration for the listener

Aviso configuration¶

CONFIG = {
    "notification_engine": {
        "host": "aviso.lumi.apps.dte.destination-earth.eu",
        "port": 443,
        "https": True,
    },
    "configuration_engine": {
        "host": "aviso.lumi.apps.dte.destination-earth.eu",
        "port": 443,
        "https": True,
    },
    "schema_parser": "generic",
    "remote_schema": True,
    "auth_type": "none",
    "quiet" : True
}  # manually defined configuration

Searching for old notifications¶

Ssearching for old notifications where available. This way users can explicitly replay past notifications and executes triggers.

import sys
START_DATE = datetime(2020, 12, 12)  # Start date for the notification listener

def triggered_function(notification):
    """
    Function for the listener to trigger.
    """
    
    #pp(notification)
    # Access the date field
    date_str = notification['request']['date']    

    # Convert the date string to a datetime object
    date_obj = datetime.strptime(date_str, '%Y%m%d')
    formatted_date = date_obj.strftime('%Y-%m-%d')
    pp("ExtremeDT data available=>" + formatted_date)

def create_hist_listener():
    """
    Creates and returns a listener configuration.
    """

    trigger = {
        "type": TRIGGER_TYPE,
        "function": triggered_function,
    }  # Define the trigger for the listener
    
    # Return the complete listener configuration
    return {"event": LISTENER_EVENT, "request": REQUEST, "triggers": [trigger]}

try:
    listener = create_hist_listener()  # Create listener configuration
    listeners_config = {"listeners": [listener]}  # Define listeners configuration
    config = user_config.UserConfig(**CONFIG)
    print("loaded config:")
    pp(CONFIG)
    nmh = NotificationManager()  # Initialize the NotificationManager

    nmh.listen(
        listeners=listeners_config, from_date=START_DATE, config=config
    )  # Start listening
except Exception as e:
    print(f"Failed to initialize the Notification Manager: {e}")

loaded config:
{'auth_type': 'none',
 'configuration_engine': {'host': 'aviso.lumi.apps.dte.destination-earth.eu',
                          'https': True,
                          'port': 443},
 'notification_engine': {'host': 'aviso.lumi.apps.dte.destination-earth.eu',
                         'https': True,
                         'port': 443},
 'quiet': True,
 'remote_schema': True,
 'schema_parser': 'generic'}