Build a fully local voice assistant in 2026 https://blog.platypush.tech/article/Local-voice-assistant

backend.http:
    port: 8008

# Enable the OpenWeatherMap plugin
weather.openweathermap:
    token: YOUR_OPENWEATHERMAP_API_KEY
    # Specify a location by name
    location: Amsterdam,NL
    # Or by coordinates
    # lat: 52.372829
    # long: 4.893680

    # How often the plugin should check for new data, in seconds
    poll_interval: 300

    # metric or imperial units
    # units: metric

docker run --rm \
    --name platypush \
    -p 8008:8008 \
    -v /your/platypush/config:/etc/platypush \
    -v /your/platypush/data:/var/lib/platypush \
    -e PLATYPUSH_DEVICE_ID=WeatherApp \
    quay.io/platypush/platypush

sudo install -d -m 700 /etc/wireguard

# Generate the server key
wg genkey | sudo tee /etc/wireguard/private.key > /dev/null
sudo chmod 600 /etc/wireguard/private.key

# Generate a public key
sudo cat /etc/wireguard/private.key | wg pubkey | sudo tee /etc/wireguard/public.key

# Print this value. You will need it in the client configuration.
sudo cat /etc/wireguard/public.key

sudo install -d -m 700 /etc/wireguard

# Generate the client key
wg genkey | sudo tee /etc/wireguard/private.key > /dev/null
sudo chmod 600 /etc/wireguard/private.key

# Generate a public key
sudo cat /etc/wireguard/private.key | wg pubkey | sudo tee /etc/wireguard/public.key

# Print this value. You will need it in the server configuration.
sudo cat /etc/wireguard/public.key

CLIENT_PUBLIC_KEY="<The public key of the client>"
SERVER_PRIVATE_KEY="$(sudo cat /etc/wireguard/private.key)"

sudo tee /etc/wireguard/wg0.conf > /dev/null <<EOF
[Interface]
PrivateKey = ${SERVER_PRIVATE_KEY}
Address = 10.0.0.1/32
ListenPort = 9929

[Peer]
PublicKey = ${CLIENT_PUBLIC_KEY}
AllowedIPs = 10.0.0.2/32
EOF

unset SERVER_PRIVATE_KEY
sudo chmod 600 /etc/wireguard/wg0.conf

sudo ufw allow 9929/udp comment 'Wireguard'

# Enable and start the tunnel
sudo systemctl enable --now wg-quick@wg0

# Verify that the tunnel is up
sudo wg show
sudo systemctl status wg-quick@wg0

SERVER_PUBLIC_KEY="<The public key of the server>"
SERVER_ENDPOINT="<The public IP or DNS name of the server>"
CLIENT_PRIVATE_KEY="$(sudo cat /etc/wireguard/private.key)"

sudo tee /etc/wireguard/wg0.conf > /dev/null <<EOF
[Interface]
PrivateKey = ${CLIENT_PRIVATE_KEY}
Address = 10.0.0.2/32

[Peer]
PublicKey = ${SERVER_PUBLIC_KEY}
AllowedIPs = 10.0.0.1/32
Endpoint = ${SERVER_ENDPOINT}:9929
PersistentKeepalive = 25
EOF

unset CLIENT_PRIVATE_KEY
sudo chmod 600 /etc/wireguard/wg0.conf

sudo systemctl enable --now wg-quick@wg0

# Verify the tunnel and the route to the reverse proxy.
sudo wg show
ping -c 3 10.0.0.1

ping -c 3 10.0.0.2
curl -I http://10.0.0.2:8008/

upstream platypush-weather {
  # Or just 127.0.0.1:8008 if the Platypush service runs on the same box
  server 10.0.0.2:8008;
}

server {
  # Replace with your own domain
  # NOTE: The DNS entry of this domain must point to the reverse proxy
  server_name weather.platypush.example.com;
  listen 80;

  # Standard HTTP resources
  location / {
      client_max_body_size 5M;
      proxy_read_timeout 60;
      proxy_connect_timeout 60;
      proxy_set_header Host $http_host;
      proxy_set_header X-Real-IP $remote_addr;
      proxy_set_header X-Forwarded-Ssl on;
      proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_pass http://platypush-weather;
  }

  # WebSockets
  location /ws/ {
      client_max_body_size 5M;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection "upgrade";
      proxy_redirect off;
      proxy_http_version 1.1;
      proxy_set_header Host $http_host;
      proxy_pass http://platypush-weather;
  }
}

sudo nginx -t
sudo nginx -s reload

curl -I http://weather.platypush.example.com/

sudo certbot --nginx -d weather.platypush.example.com

curl -I https://weather.platypush.example.com/

{
  "type": "event",
  "target": "WeatherApp",
  "origin": "WeatherApp",
  "id": "f7c26249a0b4eec5c04831fbfe11cdcd",
  "args": {
    "type": "platypush.message.event.weather.NewWeatherConditionEvent",
    "plugin_name": "weather.openweathermap",
    "summary": "Clear",
    "icon": "01d",
    "precip_intensity": 0,
    "precip_type": null,
    "temperature": 25.46,
    "apparent_temperature": 25.81,
    "humidity": 67,
    "pressure": 1020,
    "wind_speed": 3.58,
    "wind_gust": 6.71,
    "wind_direction": 344,
    "cloud_cover": 5,
    "visibility": 10000.0,
    "sunrise": "2026-07-11T03:31:24+00:00",
    "sunset": "2026-07-11T19:59:52+00:00",
    "units": "metric",
    "time": "2026-07-11T18:52:05"
  }
}

docker run \
  -v /your/ntfy/config:/etc/ntfy \
  -v /your/ntfy/cache:/var/cache/ntfy \
  -p 8080:80 \
  -it \
  binwiederhier/ntfy \
      serve \
      --cache-file /var/cache/ntfy/cache.db

ntfy:
  # Optional, if using a custom server, otherwise ntfy.sh is used
  # server_url: https://ntfy.example.com

# Content of /your/platypush/config/scripts/weather_notifications.py

import json
from functools import wraps
from time import time

from platypush import Variable, cron, run
from platypush.events.weather import NewWeatherConditionEvent

# The weather notifications topic. Replace `1234` with a random string
weather_notifications_topic = "weather-notifications-1234"

# A Platypush variable to persist the last weather notification timestamp
last_weather_state_notification_timestamp = Variable(
    "LAST_WEATHER_STATE_NOTIFICATION_TIMESTAMP"
)

# A Platypush variable to persist the latest notified weather state,
# so notifications will be delivered when the state changes, even if
# they are within the `weather_state_notification_timeout`
last_notified_weather_state = Variable("LAST_NOTIFIED_WEATHER_STATE")

# The maximum time between weather notifications
weather_state_notification_timeout = (
    60 * 60 * 60  # At most one weather notification every 30 minutes
)


def notification_throttled(func):
    """
    Prevents the decorated function from running too often, unless the
    weather state has changed since the last notification.
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        event = args[0]
        weather_state = json.dumps(event.args, sort_keys=True)
        last_weather_state = last_notified_weather_state.get()
        notification_is_throttled = (
            time() - float(last_weather_state_notification_timestamp.get() or 0)
            <= weather_state_notification_timeout
        )

        if notification_is_throttled and weather_state == last_weather_state:
            return

        result = func(*args, **kwargs)
        last_weather_state_notification_timestamp.set(time())
        last_notified_weather_state.set(weather_state)
        return result

    return wrapper


def precipitation_notification(event: NewWeatherConditionEvent) -> dict | None:
    """
    Returns a notification about precipitation for ntfy, if any.
    """

    if (event.precip_intensity or 0) < 0.1:
        return  # Negligible or absent precipitation

    precip_type = (event.precip_type or "rain").lower()
    if event.precip_intensity < 2:
        precip_intensity = "light"
        priority = 2
    elif event.precip_intensity < 5:
        precip_intensity = "moderate"
        priority = 3
    else:
        precip_intensity = "heavy"
        priority = 4

    if precip_type == "snow":
        precip_icon = "❄️"
    elif precip_type == "hail":
        precip_icon = "🌧️"
    else:
        precip_icon = "💧"

    run(
        "ntfy.send_message",
        topic=weather_notifications_topic,
    )

    return {
        "title": f"{precip_icon} Weather update",
        "message": f"{precip_intensity.capitalize()} {precip_type} in your area",
        "priority": priority,
    }


@hook(NewWeatherConditionEvent)
@notification_throttled
def on_weather_update(event: NewWeatherConditionEvent, *_, **__):
    precip = precipitation_notification(event)
    if precip:
        run(
            "ntfy.send_message",
            topic=weather_notifications_topic,
            **precip,
        )

# Content of /your/platypush/config/scripts/weather_report.py

from datetime import datetime

from platypush import cron, procedure, run

weather_plugin = "weather.openweathermap"

# The weather notifications topic. Replace `1234` with a random string
weather_notifications_topic = "weather-notifications-1234"


@procedure("deliver_weather_report")
def deliver_weather_report():
    """
    Procedure that sends a weather report to ntfy.
    """

    forecast = [
        {
            "icon": weather["icon"],
            "temperature": weather["temperature"],
            "summary": weather["summary"],
            "time": weather["time"],
        }
        for weather in run(f"{weather_plugin}.get_forecast")
        if datetime.fromisoformat(weather["time"]).day == datetime.now().day
    ]

    current_weather = {
        "time": datetime.now().isoformat(),
        **run(f"{weather_plugin}.get_current_weather"),
    }

    body = ""
    for forecast_entry in [current_weather, *forecast]:
        body += (
            f"`{datetime.fromisoformat(forecast_entry['time']).strftime('%H:%M')}`\n"
            f"**{round(forecast_entry['temperature'])}°** "
            f"![{forecast_entry['summary']}](https://openweathermap.org/payload/api/media/file/{forecast_entry['icon']}.png)\n\n"
        )

    run(
        "ntfy.send_message",
        topic=weather_notifications_topic,
        title=f"{round(current_weather['temperature'])}° - {current_weather['summary']}",
        message=body,
        icon=f"https://openweathermap.org/payload/api/media/file/{current_weather['icon']}.png",
        markdown=True,
    )


@cron("0 6 * * *")
def daily_weather_report():
    """
    Run the `deliver_weather_report` procedure every day at 6 AM.
    """
    run("procedure.deliver_weather_report")

assistant.openwakeword:
  detection_sensitivity: 0.3
  models:
    - alexa

openai:
  model: gpt-5.5
  api_key: <YOUR-OPENAI-API-KEY>

assistant.openai:
  tts_plugin: tts.openai
  conversation_start_sound: /usr/share/sounds/assistant.mp3
  input_volume: 110

tts.openai:
  output_volume: 85

# Content of /your/platypush/config/scripts/weather_assistant.py

import json
from dataclasses import dataclass
from datetime import datetime
from time import time

import requests

from platypush import Config, run, __version__ as platypush_version
from platypush.events.assistant import (
    HotwordDetectedEvent,
    SpeechRecognizedEvent,
)

ai_plugin = "openai"
assistant_plugin = "assistant.openai"
weather_plugin = "weather.openweathermap"
location_cache: dict[str, tuple[float, float]] = {}


@dataclass
class DefaultLocation:
    """
    A utility class to automatically and lazy manage the default weather
    location.
    """
    _location: str | None = None

    @property
    def name(self) -> str | None:
        if self._location:
            return self._location

        # Get the weather plugin configuration
        cfg = Config.get().get(weather_plugin)
        if not cfg:
            return None

        lat = cfg.get("lat")
        long = cfg.get("long")
        location = cfg.get("location")

        # If a location name is configured, use that
        if location:
            return location

        # Otherwise, reverse lookup the location from latitude and longitude
        self._location = (
            run(f"{weather_plugin}.reverse_lookup_location", lat=lat, long=long) or {}
        ).get("name")

        return self._location


default_location = DefaultLocation()


@dataclass
class WeatherRequest:
    """
    A weather forecast request.
    """

    location: str
    delta_days: int

    def __post_init__(self):
        if self.delta_days < 0:
            raise ValueError("delta_days must be positive")

    def get_location_coords(self) -> tuple[float, float] | None:
        """
        Get the coordinates of the location.
        """

        # Cache lookup
        coord = location_cache.get(self.location)
        if coord:
            return coord

        # Reverse geo lookup
        response = requests.get(
            f"https://nominatim.openstreetmap.org/search?q={self.location}&format=json&limit=1",
            headers={
                "User-Agent": (
                    f"Mozilla/5.0 (compatible; Platypush/{platypush_version}; "
                    "+https://git.platypush.tech/platypush/platypush)"
                ),
            },
            timeout=10,
        )

        response.raise_for_status()
        geojson = response.json() or []
        if not geojson:
            return None

        lat, lng = (geojson[0] or {}).get("lat"), (geojson[0] or {}).get("lon")
        if not (lat and lng):
            return None

        location_cache[self.location] = lat, lng
        return lat, lng

    def get_time_range(self) -> tuple[datetime, datetime]:
        return (
            datetime.fromtimestamp(time() + self.delta_days * 24 * 60 * 60),
            datetime.fromtimestamp(time() + (self.delta_days + 1) * 24 * 60 * 60),
        )


def get_structured_weather_forecast(prompt: str) -> dict:
    """
    Get a structured weather forecast given a free text prompt.
    """

    # Parse the free-text request
    request = parse_weather_request(prompt)
    response: dict = {}

    if not request:
        return response

    # Retrive the location for the weather request
    coord = request.get_location_coords()
    if not coord:
        return response

    # Get the weather forecast for the specified location and time frame
    lat, lng = coord
    time_range = request.get_time_range()
    forecast = [
        {
            "location": request.location,
            **weather,
        }
        for weather in run(
            f"{weather_plugin}.get_forecast",
            lat=lat,
            long=lng,
        )
        if time_range[0] <= datetime.fromisoformat(weather["time"]) <= time_range[1]
    ]

    # If the user requested the current weather, add it to the response
    is_today_forecast = request.delta_days == 0
    current_weather = None
    if is_today_forecast:
        current_weather = forecast[0]

    if current_weather:
        response["now"] = current_weather

    # Construct the response
    if is_today_forecast:
        key = "today"
    elif request.delta_days == 1:
        key = "tomorrow"
    else:
        key = f"{request.delta_days}days"

    response[key] = forecast
    return response


def parse_weather_request(request: str) -> WeatherRequest | None:
    """
    Parse a weather request given a free text prompt.
    """

    # Use the OpenAI plugin to parse the free-text user request into a
    # structured request
    request_dict = (
        run(
            f"{ai_plugin}.get_response",
            context=[
                {
                    "role": "system",
                    "content": (
                        "You are a voice assistant provided with weather requests as free text.\n"
                        "Given the prompt, return a structured JSON representation of the request in the following format: "
                        '{ "type": "weather", "delta_days": 1, "location": "San Francisco" }, '
                        'where both delta_days and location are optional (e.g. if the user simply asks "How\'s the weather?".\n'
                        'If the prompt doesn\'t seem to contain a weather request, return { "type": null }'
                    ),
                }
            ],
            prompt=request,
        )
        or {}
    )

    if request_dict.get("type") != "weather":
        return None

    weather_request = WeatherRequest(
        location=request_dict.get("location", default_location.name),
        delta_days=request_dict.get("delta_days", 0),
    )

    return weather_request


def get_weather_report(user_request: str) -> str | None:
    """
    Get a weather free-text report given a free text prompt.
    """

    structured_response = get_structured_weather_forecast(user_request)
    response = None

    if structured_response:
        # Use the OpenAI plugin to translate the structured weather forecast
        # response into a free-text report
        response = run(
            f"{ai_plugin}.get_response",
            prompt=json.dumps(structured_response),
            context=[
                # General weather assistant system prompt
                {
                    "role": "system",
                    "content": (
                        "You are a weather voice assistant that translates JSON weather reports "
                        "into more informal weather reports. The output reports should be brief and to the "
                        "point, but not miss relevant details from the original report."
                    ),
                },
                # JSON structure system prompt
                {
                    "role": "system",
                    "content": (
                        "If the JSON report contains a 'now' key, that's the current weather. "
                        "Otherwise, it may contain either 'today', 'tomorrow' or '<n>days' in the future. "
                    ),
                },
                # Avoid abbreviations
                {
                    "role": "system",
                    "content": (
                        "Do not use abbreviations in your output text, always use the full text units. "
                        "Also, strip any Markdown formatting from the output text. "
                        "Keep in mind that your output text will be rendered verbatim by a text-to-speech engine."
                    ),
                },
                # Degrees and wind speed system prompt
                {
                    "role": "system",
                    "content": (
                        "Do not say 'Celsius' or 'Fahreneit', only 'degrees'. "
                        "Wind speed is reported in km/h, but don't report the absolute number - "
                        "just if the wind is absent, weak, medium or strong."
                    ),
                },
                # Precipitation system prompt
                {
                    "role": "system",
                    "content": (
                        "Precipitation intensity is expressed in mm/h. Do not report the absolute number. "
                        "Only say if the precipitation is low, medium, high or very intense. "
                        "If absent, don't say anything about precipitations. "
                        "Otherwise, mention the precipitation type too (rain, snow, hail...). "
                        "If any precipitations are present in the forecast, mention their intensity and "
                        "around what time of the day they will happen."
                    ),
                },
                # Cloud cover system prompt
                {
                    "role": "system",
                    "content": (
                        "Cloud cover is reported as a percentage between 0 and 100. "
                        "Do not report the absolute number - only a description of the cloud cover. "
                        "In the forecast, mention a cloud cover description that matches a reasonable average among the data points."
                    ),
                },
            ],
        )

    return response


@when(HotwordDetectedEvent)
def on_hotword_detected():
    """
    Start the conversation when the hotword is detected.
    """
    run(f"{assistant_plugin}.start_conversation")


@when(SpeechRecognizedEvent, phrase=".*weather.*")
def on_weather_request(event):
    """
    Respond to weather requests by intercepting `SpeechRecognizedEvent` events
    that contain a weather request (regex).
    """
    response = (
        get_weather_report(event.phrase) or
        "Sorry, I couldn't find any weather information for that location."
    )

    event.assistant.render_response(response)