WIS2 Downloader Developer Guide

from shared import get_redis_client

# Get Redis client (singleton)
redis = get_redis_client()

# Use like normal Redis client
redis.set('key', 'value')
redis.get('key')

from shared import setup_logging

# Configure root logger (call once at startup)
setup_logging()

# Get module-specific logger
LOGGER = setup_logging(__name__)

LOGGER.info("Message with UTC timestamp")

def run_manager():
    # 1. Create MQTT subscriber
    mqtt_subscriber = Subscriber(**broker_config)

    # 2. Create Redis command listener
    redis_listener = CommandListener(
        subscriber=mqtt_subscriber,
        channel=COMMAND_CHANNEL
    )

    # 3. Start MQTT in separate thread
    mqtt_thread = threading.Thread(target=mqtt_subscriber.start, daemon=True)
    mqtt_thread.start()

    # 4. Start Redis listener (blocks)
    redis_listener.start()

class Subscriber:
    def __init__(self, host, port, uid, pwd, protocol, session):
        # Configure MQTT client with callbacks
        self.client = mqtt.Client(...)
        self.client.on_message = self._on_message

    def _on_message(self, client, userdata, msg):
        # Parse notification, create Celery task
        job = {
            "topic": msg.topic,
            "target": target,
            "filters": filters,
            "payload": json.loads(msg.payload)
        }
        workflow = wis2_download(job)
        workflow.apply_async()

    def subscribe(self, topic, subscriptions):
        # subscriptions: {sub_id: {id, save_path, filter}}
        self.client.subscribe(topic, qos=0)
        self.active_subscriptions[topic] = {
            "pattern": topic,
            "subscriptions": subscriptions
        }

    def add_subscription(self, topic, sub_id, save_path, filter_config):
        self.active_subscriptions[topic]["subscriptions"][sub_id] = {
            "id": sub_id, "save_path": save_path, "filter": filter_config
        }

    def remove_subscription(self, topic, sub_id):
        self.active_subscriptions[topic]["subscriptions"].pop(sub_id, None)

    def unsubscribe(self, topic):
        self.client.unsubscribe(topic)
        del self.active_subscriptions[topic]

class CommandListener(threading.Thread):
    def __init__(self, subscriber, channel):
        self.subscriber = subscriber
        self.pubsub = get_redis_client().pubsub()

    def run(self):
        self.pubsub.subscribe(self.channel)
        while not self.stop_event.is_set():
            message = self.pubsub.get_message()
            if message:
                self._process_command(message)

    def _process_command(self, message):
        command = json.loads(message['data'])
        action = command['action']
        if action == 'subscribe':
            self.subscriber.subscribe(command['topic'], command['subscriptions'])
        elif action == 'add_subscription':
            self.subscriber.add_subscription(
                command['topic'], command['sub_id'],
                command['save_path'], command['filter']
            )
        elif action == 'remove_subscription':
            self.subscriber.remove_subscription(command['topic'], command['sub_id'])
        elif action == 'unsubscribe':
            self.subscriber.unsubscribe(command['topic'])
        elif action == 'update_subscription':
            self.subscriber.add_subscription(
                command['topic'], command['sub_id'],
                command['save_path'], command['filter']
            )

@app.post('/subscriptions')
def add_subscription():
    data = get_json()
    topic = normalise_topic(data.get('topic'))
    save_path = normalise_path(data.get('target', ''))
    filter_config = data.get('filter') or data.get('filters') or {}

    sub_id = str(uuid4())
    sub_data = {'id': sub_id, 'topic': topic, 'save_path': save_path, 'filter': filter_config}

    existing = _subs_for_topic(topic, _get_all_subscriptions())
    is_new_topic = len(existing) == 0

    _persist_subscription(sub_id, sub_data)

    if is_new_topic:
        # First sub for this topic — open MQTT connection
        command = {"action": "subscribe", "topic": topic,
                   "subscriptions": {sub_id: sub_data}}
    else:
        # Topic already open — just register the new destination
        command = {"action": "add_subscription", "topic": topic,
                   "sub_id": sub_id, "save_path": save_path, "filter": filter_config}

    publish_command(command)
    return jsonify(sub_data), 201

@app.route('/metrics')
def expose_metrics():
    # Update live gauge before exposing
    queue_length = redis_client.llen(CELERY_DEFAULT_QUEUE)
    set_gauge('celery_queue_length', {'queue_name': CELERY_DEFAULT_QUEUE}, queue_length)
    return Response(generate_prometheus_text(), mimetype="text/plain")

from shared import incr_counter
incr_counter('downloads_total', {'cache': hostname, 'media_type': mime_type})

app = Celery('tasks',
             broker=CELERY_BROKER_URL,
             result_backend=CELERY_RESULT_BACKEND)

# Auto-discover tasks
app.autodiscover_tasks(['task_manager.tasks', 'task_manager.tasks.wis2'])

@app.task(bind=True)
@metrics_collector
def download_from_wis2(self, job):
    result = {...}  # Initialize result dict

    # 1. Extract identifiers
    message_id = job['payload']['id']
    data_id = job['payload']['properties']['data_id']
    filehash = job['payload']['properties']['integrity']['value']

    # 2. Deduplication check
    for key, type in [(message_id, 'by-msg-id'), ...]:
        if get_status(key, type) == STATUS_SUCCESS:
            result['status'] = STATUS_SKIPPED
            return result

    # 3. Acquire lock
    lock_acquired = redis_client.set(lock_key, 1, nx=True, ex=LOCK_EXPIRE)
    if not lock_acquired:
        raise self.retry(countdown=10, max_retries=10)

    # 4. Download file
    response = _pool.request('GET', download_url, ...)

    # 5. Pre-download filter pass (no media_type/size yet)
    pre_action, pre_reason = _apply_job_filter(job, ctx_pre)
    if pre_action == 'reject':
        result['status'] = STATUS_SKIPPED
        return result

    # 6. Download file
    response = _pool.request('GET', download_url, ...)

    # 7. Verify hash
    if hash_expected and hash_base64 != hash_expected:
        result['status'] = STATUS_FAILED
        return result

    # 8. Post-download filter pass (media_type and size now known)
    post_action, post_reason = _apply_job_filter(job, ctx_post)
    if post_action == 'reject':
        result['status'] = STATUS_SKIPPED
        return result

    # 9. Save file
    output_path.write_bytes(data)
    result['status'] = STATUS_SUCCESS
    return result

def wis2_download(args):
    workflow = download_from_wis2.s(args)
    return workflow

{
  "cache": {
    "children": {
      "a": {
        "children": {
          "wis2": {
            "children": {
              "de-dwd": {
                "children": {
                  ...
                  "synop": {
                    "datasets": [<WCMP2Record>, ...]
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

from i18n import t

# Simple lookup
label = t('nav.dashboard')                      # → 'Dashboard'

# With interpolation (uses str.format)
label = t('subscriptions.folder', path='./')    # → 'Folder: ./'

# modules/ui/i18n/de.py
"""German strings."""

STRINGS: dict[str, str] = {
    'nav.dashboard':        'Dashboard',
    'nav.catalogue':        'Katalogsuche',
    'nav.tree':             'Baumsuche',
    'nav.manual':           'Manuell abonnieren',
    'nav.manage':           'Abonnements verwalten',
    'nav.settings':         'Einstellungen',
    # ... all remaining keys ...
    'subscriptions.folder': 'Ordner: {path}',   # {path} must be preserved
    'settings.records':     '{name}: {count} Einträge',
}

from . import ar, de, en, es, fr, ru, zh        (1)

LANGUAGES: dict[str, str] = {
    'en': 'English',
    'fr': 'Français',
    'es': 'Español',
    'ar': 'العربية',
    'zh': '中文',
    'ru': 'Русский',
    'de': 'Deutsch',                             (2)
}

_STRINGS: dict[str, dict[str, str]] = {
    'en': en.STRINGS,
    'fr': fr.STRINGS,
    'es': es.STRINGS,
    'ar': ar.STRINGS,
    'zh': zh.STRINGS,
    'ru': ru.STRINGS,
    'de': de.STRINGS,                            (3)
}

RTL_LANGUAGES: frozenset[str] = frozenset({'ar', 'he'})

docker-compose restart wis2downloader-ui

# modules/ui/i18n/en.py
'myview.title':       'My New View',
'myview.description': 'Showing results for {topic}.',

# fr.py / es.py / ar.py / zh.py / ru.py
'myview.title':       'My New View',         # TODO: translate
'myview.description': 'Showing results for {topic}.',

from i18n import t

def render(container):
    with container:
        ui.label(t('myview.title')).classes('page-title')
        ui.label(t('myview.description', topic=selected_topic))