diff options
| author | Martin Guzman <55927935+brockar@users.noreply.github.com> | 2025-11-22 13:42:39 -0300 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2025-11-22 13:42:39 -0300 |
| commit | bcfd5e7c2b67f0a62b1ceeb62d20b0a80ca55a70 (patch) | |
| tree | e87f649fe0c9d8f382b8f465d1468b6495fb5886 /config/hypr/UserScripts | |
| parent | c862835221decf03b6640bb43e4044861cceaa04 (diff) | |
| parent | 1351220a092414ae593bc006e3a3ebc09523198d (diff) | |
Merge pull request #871 from JaKooLit/development
[Bug] Change to waybar sleep timers was too long, slowed down waybar start/refressh
Diffstat (limited to 'config/hypr/UserScripts')
| -rwxr-xr-x | config/hypr/UserScripts/RainbowBorders.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/RofiBeats.sh | 203 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/RofiCalc.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/Tak0-Autodispatch.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/WallpaperAutoChange.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/WallpaperEffects.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/WallpaperRandom.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/WallpaperSelect.sh | 2 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/Weather.py | 767 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/Weather.sh | 164 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/WeatherWrap.sh | 33 | ||||
| -rwxr-xr-x | config/hypr/UserScripts/ZshChangeTheme.sh | 2 |
12 files changed, 856 insertions, 327 deletions
diff --git a/config/hypr/UserScripts/RainbowBorders.sh b/config/hypr/UserScripts/RainbowBorders.sh index cc1419fb..0a7fd721 100755 --- a/config/hypr/UserScripts/RainbowBorders.sh +++ b/config/hypr/UserScripts/RainbowBorders.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## # for rainbow borders animation diff --git a/config/hypr/UserScripts/RofiBeats.sh b/config/hypr/UserScripts/RofiBeats.sh index 1cddce09..a002a518 100755 --- a/config/hypr/UserScripts/RofiBeats.sh +++ b/config/hypr/UserScripts/RofiBeats.sh @@ -1,35 +1,39 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## -# For Rofi Beats to play online Music or Locally saved media files +# RofiBeats - unified, dynamic UI (add, remove, manage, play) -# Variables mDIR="$HOME/Music/" iDIR="$HOME/.config/swaync/icons" rofi_theme="$HOME/.config/rofi/config-rofi-Beats.rasi" -rofi_theme_1="$HOME/.config/rofi/config-rofi-Beats-menu.rasi" +rofi_theme_menu="$HOME/.config/rofi/config-rofi-Beats-menu.rasi" +music_list="$HOME/.config/rofi/online_music.list" -# Online Stations. Edit as required -declare -A online_music=( - ["FM - Easy Rock 96.3 ๐ป๐ถ"]="https://radio-stations-philippines.com/easy-rock" - ["FM - Easy Rock - Baguio 91.9 ๐ป๐ถ"]="https://radio-stations-philippines.com/easy-rock-baguio" - ["FM - Love Radio 90.7 ๐ป๐ถ"]="https://radio-stations-philippines.com/love" - ["FM - WRock - CEBU 96.3 ๐ป๐ถ"]="https://onlineradio.ph/126-96-3-wrock.html" - ["FM - Fresh Philippines ๐ป๐ถ"]="https://onlineradio.ph/553-fresh-fm.html" - ["Radio - Lofi Girl ๐ง๐ถ"]="https://play.streamafrica.net/lofiradio" - ["Radio - Chillhop ๐ง๐ถ"]="http://stream.zeno.fm/fyn8eh3h5f8uv" - ["Radio - Ibiza Global ๐ง๐ถ"]="https://filtermusic.net/ibiza-global" - ["Radio - Metal Music ๐ง๐ถ"]="https://tunein.com/radio/mETaLmuSicRaDio-s119867/" - ["YT - Wish 107.5 YT Pinoy HipHop ๐ป๐ถ"]="https://youtube.com/playlist?list=PLkrzfEDjeYJnmgMYwCKid4XIFqUKBVWEs&si=vahW_noh4UDJ5d37" - ["YT - Youtube Top 100 Songs Global ๐น๐ถ"]="https://youtube.com/playlist?list=PL4fGSI1pDJn6puJdseH2Rt9sMvt9E2M4i&si=5jsyfqcoUXBCSLeu" - ["YT - Wish 107.5 YT Wishclusives ๐น๐ถ"]="https://youtube.com/playlist?list=PLkrzfEDjeYJn5B22H9HOWP3Kxxs-DkPSM&si=d_Ld2OKhGvpH48WO" - ["YT - Relaxing Piano Music ๐น๐ถ"]="https://youtu.be/6H7hXzjFoVU?si=nZTPREC9lnK1JJUG" - ["YT - Youtube Remix ๐น๐ถ"]="https://youtube.com/playlist?list=PLeqTkIUlrZXlSNn3tcXAa-zbo95j0iN-0" - ["YT - Korean Drama OST ๐น๐ถ"]="https://youtube.com/playlist?list=PLUge_o9AIFp4HuA-A3e3ZqENh63LuRRlQ" - ["YT - lofi hip hop radio beats ๐น๐ถ"]="https://www.youtube.com/live/jfKfPfyJRdk?si=PnJIA9ErQIAw6-qd" - ["YT - Relaxing Piano Jazz Music ๐น๐ถ"]="https://youtu.be/85UEqRat6E4?si=jXQL1Yp2VP_G6NSn" -) +mkdir -p "$(dirname "$music_list")" +[[ -f "$music_list" ]] || touch "$music_list" -# Populate local_music array with files from music directory and subdirectories +# Send notification +notification() { + notify-send -u normal -i "$iDIR/music.png" "$@" +} + +# Check if mpv is currently playing +music_playing() { pgrep -x "mpv" >/dev/null; } + +# Stop all mpv processes except mpvpaper +stop_music() { + mpv_pids=$(pgrep -x mpv) + if [ -n "$mpv_pids" ]; then + mpvpaper_pid=$(ps aux | grep -- 'unique-wallpaper-process' | grep -v 'grep' | awk '{print $2}') + for pid in $mpv_pids; do + if ! echo "$mpvpaper_pid" | grep -q "$pid"; then + kill -9 $pid || true + fi + done + notification "Music stopped" + fi +} + +# Populate local music file list populate_local_music() { local_music=() filenames=() @@ -39,115 +43,94 @@ populate_local_music() { done < <(find -L "$mDIR" -type f \( -iname "*.mp3" -o -iname "*.flac" -o -iname "*.wav" -o -iname "*.ogg" -o -iname "*.mp4" \)) } -# Function for displaying notifications -notification() { - notify-send -u normal -i "$iDIR/music.png" "Now Playing:" "$@" -} - -# Main function for playing local music +# Play selected local music file play_local_music() { populate_local_music - - # Prompt the user to select a song - choice=$(printf "%s\n" "${filenames[@]}" | rofi -i -dmenu -config $rofi_theme) - - if [ -z "$choice" ]; then - exit 1 - fi - - # Find the corresponding file path based on user's choice and set that to play the song then continue on the list - for (( i=0; i<"${#filenames[@]}"; ++i )); do + choice=$(printf "%s\n" "${filenames[@]}" | rofi -i -dmenu -config "$rofi_theme" \ + -theme-str 'entry { placeholder: "๐ต Choose Local Music"; }') + [[ -z "$choice" ]] && exit 1 + for ((i = 0; i < "${#filenames[@]}"; ++i)); do if [ "${filenames[$i]}" = "$choice" ]; then - - if music_playing; then - stop_music - fi - notification "$choice" - mpv --playlist-start="$i" --loop-playlist --vid=no "${local_music[@]}" - + music_playing && stop_music + notification "Now Playing:" "$choice" + mpv --no-video --playlist-start="$i" --loop-playlist "${local_music[@]}" break fi done } -# Main function for shuffling local music +# Shuffle and play all local music shuffle_local_music() { - if music_playing; then - stop_music - fi + music_playing && stop_music notification "Shuffle Play local music" - - # Play music in $mDIR on shuffle - mpv --shuffle --loop-playlist --vid=no "$mDIR" + mpv --no-video --shuffle --loop-playlist "$mDIR" } -# Main function for playing online music +# Play selected online music play_online_music() { - choice=$(for online in "${!online_music[@]}"; do - echo "$online" - done | sort | rofi -i -dmenu -config "$rofi_theme") - - if [ -z "$choice" ]; then - exit 1 + if [ ! -s "$music_list" ]; then + notify-send -u low -i "$iDIR/music.png" "No online music found" "Add some with Manage Music" + exit 0 fi - - link="${online_music[$choice]}" - - if music_playing; then - stop_music - fi - notification "$choice" - - # Play the selected online music using mpv - mpv --shuffle --vid=no "$link" -} - -# Function to check if music is already playing -music_playing() { - pgrep -x "mpv" > /dev/null + choice=$(awk -F'|' '{print $1}' "$music_list" | sort | rofi -i -dmenu -config "$rofi_theme" \ + -theme-str 'entry { placeholder: "๐ Choose Online Station"; }') + [[ -z "$choice" ]] && exit 1 + link=$(awk -F'|' -v name="$choice" '$1 == name {print $2; exit}' "$music_list") + [[ -z "$link" ]] && { + notify-send -u low -i "$iDIR/music.png" "URL not found for" "$choice" + exit 1 + } + music_playing && stop_music + notification "Now Playing:" "$choice" + mpv --no-video --shuffle "$link" } -# Function to stop music and kill mpv processes -stop_music() { - mpv_pids=$(pgrep -x mpv) +# Manage online music list (add, remove, view) +manage_music() { + sub_choice=$(printf "Add Music\nRemove Music\nView List" | rofi -dmenu \ + -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐ ๏ธ Manage Music List"; }') - if [ -n "$mpv_pids" ]; then - # Get the PID of the mpv process used by mpvpaper (using the unique argument added) - mpvpaper_pid=$(ps aux | grep -- 'unique-wallpaper-process' | grep -v 'grep' | awk '{print $2}') - - for pid in $mpv_pids; do - if ! echo "$mpvpaper_pid" | grep -q "$pid"; then - kill -9 $pid || true - fi - done - notify-send -u low -i "$iDIR/music.png" "Music stopped" || true - fi + case "$sub_choice" in + "Add Music") + name=$(rofi -dmenu -lines 0 -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐ผ Enter Music Title"; }') + [[ -z "$name" ]] && return + url=$(rofi -dmenu -lines 0 -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐ Enter Music URL"; }') + [[ -z "$url" ]] && return + echo "$name|$url" >>"$music_list" + notification "Added" "$name" + ;; + "Remove Music") + entry=$(awk -F'|' '{print $1}' "$music_list" | rofi -dmenu -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐๏ธ Select Music to Remove"; }') + [[ -z "$entry" ]] && return + grep -vF "$entry" "$music_list" >"$music_list.tmp" && mv "$music_list.tmp" "$music_list" + notification "Removed" "$entry" + ;; + "View List") + # Show only titles, not URLs + awk -F'|' '{print $1}' "$music_list" | rofi -dmenu -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐ Online Music List"; }' >/dev/null + ;; + esac } +# Main menu user_choice=$(printf "%s\n" \ "Play from Online Stations" \ "Play from Music directory" \ "Shuffle Play from Music directory" \ "Stop RofiBeats" \ - | rofi -dmenu -config $rofi_theme_1) - -echo "User choice: $user_choice" + "Manage Music List" | + rofi -dmenu -config "$rofi_theme_menu" \ + -theme-str 'entry { placeholder: "๐ง RofiBeats Menu"; }') case "$user_choice" in - "Play from Online Stations") - play_online_music - ;; - "Play from Music directory") - play_local_music - ;; - "Shuffle Play from Music directory") - shuffle_local_music - ;; - "Stop RofiBeats") - if music_playing; then - stop_music - fi - ;; - *) - ;; +"Play from Online Stations") play_online_music ;; +"Play from Music directory") play_local_music ;; +"Shuffle Play from Music directory") shuffle_local_music ;; +"Stop RofiBeats") music_playing && stop_music ;; +"Manage Music List") manage_music ;; esac diff --git a/config/hypr/UserScripts/RofiCalc.sh b/config/hypr/UserScripts/RofiCalc.sh index 4b3b8b69..b72d5f3e 100755 --- a/config/hypr/UserScripts/RofiCalc.sh +++ b/config/hypr/UserScripts/RofiCalc.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ # /* Calculator (using qalculate) and rofi */ # /* Submitted by: https://github.com/JosephArmas */ diff --git a/config/hypr/UserScripts/Tak0-Autodispatch.sh b/config/hypr/UserScripts/Tak0-Autodispatch.sh index a1f72129..114a3e8e 100755 --- a/config/hypr/UserScripts/Tak0-Autodispatch.sh +++ b/config/hypr/UserScripts/Tak0-Autodispatch.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # USAGE / ะะะกะขะ ะฃะะฆะะฏ: # 1) Run from terminal: # ./dispatch.sh <application_command> <target_workspace_number> diff --git a/config/hypr/UserScripts/WallpaperAutoChange.sh b/config/hypr/UserScripts/WallpaperAutoChange.sh index a6d2cedd..6d8e8735 100755 --- a/config/hypr/UserScripts/WallpaperAutoChange.sh +++ b/config/hypr/UserScripts/WallpaperAutoChange.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## # source https://wiki.archlinux.org/title/Hyprland#Using_a_script_to_change_wallpaper_every_X_minutes diff --git a/config/hypr/UserScripts/WallpaperEffects.sh b/config/hypr/UserScripts/WallpaperEffects.sh index ac8fc0e8..89577efa 100755 --- a/config/hypr/UserScripts/WallpaperEffects.sh +++ b/config/hypr/UserScripts/WallpaperEffects.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ # # Wallpaper Effects using ImageMagick (SUPER SHIFT W) diff --git a/config/hypr/UserScripts/WallpaperRandom.sh b/config/hypr/UserScripts/WallpaperRandom.sh index 79396508..654d4bd3 100755 --- a/config/hypr/UserScripts/WallpaperRandom.sh +++ b/config/hypr/UserScripts/WallpaperRandom.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## # Script for Random Wallpaper ( CTRL ALT W) diff --git a/config/hypr/UserScripts/WallpaperSelect.sh b/config/hypr/UserScripts/WallpaperSelect.sh index 466832ba..9e51125f 100755 --- a/config/hypr/UserScripts/WallpaperSelect.sh +++ b/config/hypr/UserScripts/WallpaperSelect.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ # This script for selecting wallpapers (SUPER W) diff --git a/config/hypr/UserScripts/Weather.py b/config/hypr/UserScripts/Weather.py index ca1d5281..a6483777 100755 --- a/config/hypr/UserScripts/Weather.py +++ b/config/hypr/UserScripts/Weather.py @@ -3,21 +3,48 @@ # Rewritten to use Open-Meteo APIs (worldwide, no API key) for robust weather data. # Outputs Waybar-compatible JSON and a simple text cache. +from __future__ import annotations + import json import os import sys import time import html -from typing import Any, Dict, List, Optional, Tuple - +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, TypeVar, Union, cast +from typing import NamedTuple import requests +from dataclasses import dataclass + +@dataclass +class Location: + lat: float + lon: float + place: Optional[str] = None + + +@dataclass +class WeatherData: + temp_str: str + feels_str: str + icon: str + status: str + min_max: str + wind_text: str + humidity_text: str + visibility_text: str + aqi_text: str + hourly_precip: str + is_day: int + code: int + # =============== Configuration =============== # You can configure behavior via environment variables OR the constants below. # Examples (zsh): # # One-off run # # WEATHER_UNITS can be "metric" or "imperial" -# WEATHER_UNITS=imperial WEATHER_PLACE="Concord, NH" python3 /home/dwilliams/Projects/Weather.py +# WEATHER_UNITS=imperial WEATHER_PLACE="Concord, NH" python3 ~/.config/hypr/UserScripts/Weather.py # # # Persist in current shell session # export WEATHER_UNITS=imperial @@ -27,10 +54,10 @@ import requests # export WEATHER_TOOLTIP_MARKUP=1 # 1 to enable Pango markup, 0 to disable # export WEATHER_LOC_ICON="๐" # or "*" for ASCII-only # -CACHE_DIR = os.path.expanduser("~/.cache") -API_CACHE_PATH = os.path.join(CACHE_DIR, "open_meteo_cache.json") -SIMPLE_TEXT_CACHE_PATH = os.path.join(CACHE_DIR, ".weather_cache") -CACHE_TTL_SECONDS = int(os.getenv("WEATHER_CACHE_TTL", "600")) # default 10 minutes +CACHE_DIR: Path = Path.home() / ".cache" +API_CACHE_PATH: Path = CACHE_DIR / "open_meteo_cache.json" +SIMPLE_TEXT_CACHE_PATH: Path = CACHE_DIR / ".weather_cache" +CACHE_TTL_SECONDS = int(os.getenv("WEATHER_CACHE_TTL", "300")) # default 5 minutes # Units: metric or imperial (default metric) UNITS = os.getenv("WEATHER_UNITS", "metric").strip().lower() # metric|imperial @@ -38,16 +65,17 @@ UNITS = os.getenv("WEATHER_UNITS", "metric").strip().lower() # metric|imperial # Optional manual coordinates ENV_LAT = os.getenv("WEATHER_LAT") ENV_LON = os.getenv("WEATHER_LON") -# Optional manual place override for tooltip +# Optional manual place override for tooltip (and optional forward geocoding) ENV_PLACE = os.getenv("WEATHER_PLACE") -# Manual place name set inside this file. If set (non-empty), this takes top priority. +# Manual place name set inside this file. If set (non-empty), this takes top priority for display +# and, if coordinates are not provided, will be used to geocode latitude/longitude. # Example: MANUAL_PLACE = "Concord, NH, US" -MANUAL_PLACE: Optional[str] = None +MANUAL_PLACE: Optional[str] = "" #Set your city HERE # Location icon in tooltip (default to a standard emoji to avoid missing glyphs) LOC_ICON = os.getenv("WEATHER_LOC_ICON", "๐") # Enable/disable Pango markup in tooltip (1/0, true/false) -TOOLTIP_MARKUP = os.getenv("WEATHER_TOOLTIP_MARKUP", "1").lower() not in ("0", "false", "no") +TOOLTIP_MARKUP = os.getenv("WEATHER_TOOLTIP_MARKUP", "0").lower() in ("1", "true", "yes") # Optional debug logging to stderr (set WEATHER_DEBUG=1 to enable) DEBUG = os.getenv("WEATHER_DEBUG", "0").lower() not in ("0", "false", "no") @@ -138,19 +166,68 @@ def log_debug(msg: str) -> None: def ensure_cache_dir() -> None: try: - os.makedirs(CACHE_DIR, exist_ok=True) + # CACHE_DIR is a Path + CACHE_DIR.mkdir(parents=True, exist_ok=True) except Exception as e: print(f"Error creating cache dir: {e}", file=sys.stderr) +def _coerce_numeric(value: Any, to_int: bool) -> Optional[Union[int, float]]: + if to_int: + if isinstance(value, int): + return value + if isinstance(value, (float, str)): + try: + return int(float(value)) + except (ValueError, TypeError): + return None + return None + else: + if isinstance(value, float): + return value + if isinstance(value, int): + return float(value) + if isinstance(value, str): + try: + return float(value) + except (ValueError, TypeError): + return None + return None + + +def coerce_int(value: Any) -> Optional[int]: + return cast(Optional[int], _coerce_numeric(value, True)) + + +def coerce_float(value: Any) -> Optional[float]: + return cast(Optional[float], _coerce_numeric(value, False)) + + +def coerce_number(value: Any) -> Union[int, float, None]: + if isinstance(value, (int, float)): + return value + if isinstance(value, str): + try: + # Parse to float, then return int if it has no fractional part + f = float(value) + return int(f) if f.is_integer() else f + except (ValueError, TypeError): + return None + return None + + def read_api_cache() -> Optional[Dict[str, Any]]: try: - if not os.path.exists(API_CACHE_PATH): + if not API_CACHE_PATH.exists(): return None - with open(API_CACHE_PATH, "r", encoding="utf-8") as f: + with API_CACHE_PATH.open("r", encoding="utf-8") as f: data = json.load(f) - if (time.time() - data.get("timestamp", 0)) <= CACHE_TTL_SECONDS: - return data + # Use ensure_dict for safety + data_dict = ensure_dict(data) + timestamp_val = data_dict.get("timestamp", 0) + timestamp = coerce_float(timestamp_val) or 0 + if (time.time() - timestamp) <= CACHE_TTL_SECONDS: + return data_dict return None except Exception as e: print(f"Error reading cache: {e}", file=sys.stderr) @@ -161,7 +238,7 @@ def write_api_cache(payload: Dict[str, Any]) -> None: try: ensure_cache_dir() payload["timestamp"] = time.time() - with open(API_CACHE_PATH, "w", encoding="utf-8") as f: + with API_CACHE_PATH.open("w", encoding="utf-8") as f: json.dump(payload, f) except Exception as e: print(f"Error writing API cache: {e}", file=sys.stderr) @@ -170,34 +247,42 @@ def write_api_cache(payload: Dict[str, Any]) -> None: def write_simple_text_cache(text: str) -> None: try: ensure_cache_dir() - with open(SIMPLE_TEXT_CACHE_PATH, "w", encoding="utf-8") as f: + with SIMPLE_TEXT_CACHE_PATH.open("w", encoding="utf-8") as f: f.write(text) except Exception as e: print(f"Error writing simple cache: {e}", file=sys.stderr) -def get_coords() -> Tuple[float, float]: - # 1) Explicit env +def get_coords_from_env() -> Optional[Tuple[float, float]]: if ENV_LAT and ENV_LON: try: return float(ENV_LAT), float(ENV_LON) except ValueError: print("Invalid WEATHER_LAT/WEATHER_LON; falling back to IP geolocation", file=sys.stderr) + return None + - # 2) Try cached coordinates from last successful forecast +def get_coords_from_cache() -> Optional[Tuple[float, float]]: try: cached = read_api_cache() - if cached and isinstance(cached, dict): - fc = cached.get("forecast") or {} - lat = fc.get("latitude") - lon = fc.get("longitude") - if isinstance(lat, (int, float)) and isinstance(lon, (int, float)): - return float(lat), float(lon) + if cached: + fc = ensure_dict(cached.get("forecast")) + lat_raw = safe_get(fc, "latitude") + lon_raw = safe_get(fc, "longitude") + lat = coerce_float(lat_raw) + lon = coerce_float(lon_raw) + if lat is None: + log_debug(f"Unexpected type for cached latitude: {type(lat_raw)}") + if lon is None: + log_debug(f"Unexpected type for cached longitude: {type(lon_raw)}") + if lat is not None and lon is not None: + return lat, lon except Exception as e: print(f"Reading cached coords failed: {e}", file=sys.stderr) + return None - # 3) IP-based geolocation with multiple providers (prefer ipwho.is, ipapi.co; ipinfo.io as fallback) - # ipwho.is + +def get_coords_from_ipwho() -> Optional[Tuple[float, float]]: try: resp = SESSION.get("https://ipwho.is/", timeout=TIMEOUT) resp.raise_for_status() @@ -209,8 +294,10 @@ def get_coords() -> Tuple[float, float]: return float(lat), float(lon) except Exception as e: print(f"ipwho.is failed: {e}", file=sys.stderr) + return None - # ipapi.co + +def get_coords_from_ipapi() -> Optional[Tuple[float, float]]: try: resp = SESSION.get("https://ipapi.co/json", timeout=TIMEOUT) resp.raise_for_status() @@ -221,8 +308,10 @@ def get_coords() -> Tuple[float, float]: return float(lat), float(lon) except Exception as e: print(f"ipapi.co failed: {e}", file=sys.stderr) + return None + - # ipinfo.io (fallback) +def get_coords_from_ipinfo() -> Optional[Tuple[float, float]]: try: resp = SESSION.get("https://ipinfo.io/json", timeout=TIMEOUT) resp.raise_for_status() @@ -233,8 +322,68 @@ def get_coords() -> Tuple[float, float]: return float(lat_s), float(lon_s) except Exception as e: print(f"ipinfo.io failed: {e}", file=sys.stderr) + return None + - # 4) Last resort +def get_coords_from_place_name(name: str) -> Optional[Tuple[float, float]]: + """Forward geocode a place name to coordinates using Open-Meteo Geocoding API. + + Returns (lat, lon) if found, else None. + """ + try: + base = "https://geocoding-api.open-meteo.com/v1/search" + params: Dict[str, Union[str, float]] = { + "name": name, + "count": 1, + "language": os.getenv("WEATHER_LANG", "en"), + "format": "json", + } + resp = SESSION.get(base, params=params, timeout=TIMEOUT) + resp.raise_for_status() + data = ensure_dict(resp.json()) + results = ensure_list(data.get("results")) + if results: + p = ensure_dict(results[0]) + lat = coerce_float(p.get("latitude")) + lon = coerce_float(p.get("longitude")) + if lat is not None and lon is not None: + return float(lat), float(lon) + except Exception as e: + print(f"Place geocoding failed: {e}", file=sys.stderr) + return None + + +def get_coords() -> Tuple[float, float]: + # 1) Forward geocode from MANUAL_PLACE first (highest priority) + if MANUAL_PLACE: + place_name = MANUAL_PLACE.strip() + coords = get_coords_from_place_name(place_name) + if coords: + return coords + + # 2) Explicit env coordinates + coords = get_coords_from_env() + if coords: + return coords + + # 3) Forward geocode from ENV_PLACE + if ENV_PLACE: + place_name = ENV_PLACE.strip() + coords = get_coords_from_place_name(place_name) + if coords: + return coords + + # 4) Try cached coordinates + coords = get_coords_from_cache() + if coords: + return coords + + # 5) IP-based geolocation + coords = get_coords_from_ipwho() or get_coords_from_ipapi() or get_coords_from_ipinfo() + if coords: + return coords + + # 6) Last resort print("IP geolocation failed: no providers succeeded", file=sys.stderr) return 0.0, 0.0 @@ -272,7 +421,7 @@ def format_visibility(meters: Optional[float]) -> str: def fetch_open_meteo(lat: float, lon: float) -> Dict[str, Any]: base = "https://api.open-meteo.com/v1/forecast" - params = { + params: Dict[str, Union[str, float]] = { "latitude": lat, "longitude": lon, "current": "temperature_2m,apparent_temperature,relative_humidity_2m,wind_speed_10m,wind_direction_10m,weather_code,visibility,precipitation,pressure_msl,is_day", @@ -289,7 +438,7 @@ def fetch_open_meteo(lat: float, lon: float) -> Dict[str, Any]: def fetch_aqi(lat: float, lon: float) -> Optional[Dict[str, Any]]: try: base = "https://air-quality-api.open-meteo.com/v1/air-quality" - params = { + params: Dict[str, Union[str, float]] = { "latitude": lat, "longitude": lon, "current": "european_aqi", @@ -303,37 +452,54 @@ def fetch_aqi(lat: float, lon: float) -> Optional[Dict[str, Any]]: return None -def fetch_place(lat: float, lon: float) -> Optional[str]: - """Reverse geocode lat/lon to an approximate place. Tries Nominatim first, then Open-Meteo.""" - lang = os.getenv("WEATHER_LANG", "en") +def extract_place_parts_nominatim(data_dict: JSONDict) -> List[str]: + address = ensure_dict(data_dict.get("address")) + candidates = [data_dict.get("name"), address.get("city"), address.get("town"), address.get("village"), address.get("hamlet")] + name = cast(Optional[str], next((c for c in candidates if c is not None), None)) + admin1 = cast(Optional[str], address.get("state")) + country = cast(Optional[str], address.get("country")) + parts: List[str] = [] + if name is not None: + parts.append(name) + if admin1 is not None: + parts.append(admin1) + if country is not None: + parts.append(country) + return parts - # 1) Nominatim (OpenStreetMap) + +def extract_place_parts_open_meteo(p: JSONDict) -> List[str]: + name = cast(Optional[str], p.get("name")) + admin1 = cast(Optional[str], p.get("admin1")) + country = cast(Optional[str], p.get("country")) + parts: List[str] = [] + if name is not None: + parts.append(name) + if admin1 is not None: + parts.append(admin1) + if country is not None: + parts.append(country) + return parts + + +def reverse_geocode(base: str, params: Dict[str, Union[str, float]], headers: Optional[Dict[str, str]] = None) -> Optional[str]: try: - base = "https://nominatim.openstreetmap.org/reverse" - params = { - "lat": lat, - "lon": lon, - "format": "jsonv2", - "accept-language": lang, - } - headers = {"User-Agent": UA + " Weather.py/1.0"} resp = SESSION.get(base, params=params, headers=headers, timeout=TIMEOUT) resp.raise_for_status() data = resp.json() - address = data.get("address", {}) - name = data.get("name") or address.get("city") or address.get("town") or address.get("village") or address.get("hamlet") - admin1 = address.get("state") - country = address.get("country") - parts = [part for part in [name, admin1, country] if part] + data_dict = ensure_dict(data) + parts = extract_place_parts_nominatim(data_dict) if parts: return ", ".join(parts) except Exception as e: - log_debug(f"Reverse geocoding (Nominatim) failed: {e}") + log_debug(f"Reverse geocoding failed: {e}") + return None - # 2) Open-Meteo reverse (fallback) + +def reverse_geocode_open_meteo(lat: float, lon: float, lang: str) -> Optional[str]: try: base = "https://geocoding-api.open-meteo.com/v1/reverse" - params = { + params: Dict[str, Union[str, float]] = { "latitude": lat, "longitude": lon, "language": lang, @@ -342,48 +508,117 @@ def fetch_place(lat: float, lon: float) -> Optional[str]: resp = SESSION.get(base, params=params, timeout=TIMEOUT) resp.raise_for_status() data = resp.json() - results = data.get("results") or [] + data_dict = ensure_dict(data) + results = ensure_list(data_dict.get("results")) if results: - p = results[0] - name = p.get("name") - admin1 = p.get("admin1") - country = p.get("country") - parts = [part for part in [name, admin1, country] if part] + p = ensure_dict(results[0]) + parts = extract_place_parts_open_meteo(p) if parts: return ", ".join(parts) except Exception as e: log_debug(f"Reverse geocoding (Open-Meteo) failed: {e}") - return None +def fetch_place(lat: float, lon: float) -> Optional[str]: + """Reverse geocode lat/lon to an approximate place. Tries Nominatim first, then Open-Meteo.""" + lang = os.getenv("WEATHER_LANG", "en") + + # 1) Nominatim (OpenStreetMap) + base = "https://nominatim.openstreetmap.org/reverse" + params: Dict[str, Union[str, float]] = { + "lat": lat, + "lon": lon, + "format": "jsonv2", + "accept-language": lang, + } + headers = {"User-Agent": UA + " Weather.py/1.0"} + place = reverse_geocode(base, params, headers) + if place: + return place + + # 2) Open-Meteo reverse (fallback) + return reverse_geocode_open_meteo(lat, lon, lang) + + # =============== Build Output =============== -def safe_get(dct: Dict[str, Any], *keys, default=None): - cur: Any = dct - for k in keys: +_T = TypeVar("_T") + +JSONValue = Union[str, int, float, bool, None, "JSONDict", "JSONList"] +JSONDict = Dict[str, JSONValue] +JSONList = List[JSONValue] + + +def ensure_dict(value: Any) -> JSONDict: + """Return a JSON-like dict when the incoming value looks like one.""" + if isinstance(value, dict): + return cast(JSONDict, value) + # Warn about unexpected type to catch API shape mismatches + val_repr = repr(value) if value is not None else "None" + if len(val_repr) > 100: + val_repr = val_repr[:100] + "..." + print(f"Warning: ensure_dict received {type(value).__name__} instead of dict: {val_repr}", file=sys.stderr) + return cast(JSONDict, {}) + + +def ensure_list(value: Any) -> JSONList: + """Return a JSON-like list when the incoming value looks like one.""" + if isinstance(value, list): + return cast(JSONList, value) + # Warn about unexpected type to catch API shape mismatches + val_repr = repr(value) if value is not None else "None" + if len(val_repr) > 100: + val_repr = val_repr[:100] + "..." + print(f"Warning: ensure_list received {type(value).__name__} instead of list: {val_repr}", file=sys.stderr) + return cast(JSONList, []) + + +def safe_get( + obj: JSONValue | None, + *keys: Union[str, int], + default: _T | None = None, +) -> _T | JSONValue | None: + """Safely traverse nested dict/list structures. + + Keys may be strings (for mapping lookups) or ints (for list indices). + Returns ``default`` if any lookup fails. + """ + + cur: JSONValue | None = obj + for key in keys: if isinstance(cur, dict): - if k not in cur: + if not isinstance(key, str) or key not in cur: return default - cur = cur[k] + cur = cur[key] elif isinstance(cur, list): - try: - cur = cur[k] # type: ignore[index] - except Exception: + if not isinstance(key, int) or key < 0 or key >= len(cur): return default + cur = cur[key] else: return default - return cur + return cast(_T | JSONValue | None, cur) + + +def get_precipitation_probabilities(forecast: JSONDict) -> List[Optional[float]]: + probs_raw = safe_get(forecast, "hourly", "precipitation_probability") + probs_raw_list = ensure_list(probs_raw) + return [coerce_float(p) if p is not None else None for p in probs_raw_list] + + +def find_current_index(times: List[str], cur_time: Optional[str]) -> int: + if cur_time is not None and cur_time in times: + return times.index(cur_time) + return 0 -def build_hourly_precip(forecast: Dict[str, Any]) -> str: +def build_hourly_precip(forecast: JSONDict) -> str: try: - times: List[str] = safe_get(forecast, "hourly", "time", default=[]) or [] - probs: List[Optional[float]] = safe_get( - forecast, "hourly", "precipitation_probability", default=[] - ) or [] - cur_time: Optional[str] = safe_get(forecast, "current", "time") - idx = times.index(cur_time) if cur_time in times else 0 + times_raw = safe_get(forecast, "hourly", "time") + times: List[str] = cast(List[str], ensure_list(times_raw)) + probs = get_precipitation_probabilities(forecast) + cur_time: Optional[str] = cast(Optional[str], safe_get(forecast, "current", "time")) + idx = find_current_index(times, cur_time) window = probs[idx : idx + 6] if not window: return "" @@ -393,152 +628,308 @@ def build_hourly_precip(forecast: Dict[str, Any]) -> str: return "" -def build_output(lat: float, lon: float, forecast: Dict[str, Any], aqi: Optional[Dict[str, Any]], place: Optional[str] = None) -> Tuple[Dict[str, Any], str]: - cur = forecast.get("current", {}) - cur_units = forecast.get("current_units", {}) - daily = forecast.get("daily", {}) - daily_units = forecast.get("daily_units", {}) +def build_weather_strings(cur: JSONDict, cur_units: JSONDict, daily: JSONDict, daily_units: JSONDict, temp_unit: str) -> Tuple[str, str, int, int, str, str, str]: + temp_val = coerce_float(cur.get("temperature_2m")) + temp_unit_str = cast(str, cur_units.get("temperature_2m", "")) + temp_str = f"{int(round(temp_val))}{temp_unit_str}" if temp_val is not None else "N/A" - temp_val = cur.get("temperature_2m") - temp_unit = cur_units.get("temperature_2m", "") - temp_str = f"{int(round(temp_val))}{temp_unit}" if isinstance(temp_val, (int, float)) else "N/A" + feels_val = coerce_float(cur.get("apparent_temperature")) + feels_unit = cast(str, cur_units.get("apparent_temperature", "")) + feels_str = f"Feels like {int(round(feels_val))}{feels_unit}" if feels_val is not None else "" - feels_val = cur.get("apparent_temperature") - feels_unit = cur_units.get("apparent_temperature", "") - feels_str = f"Feels like {int(round(feels_val))}{feels_unit}" if isinstance(feels_val, (int, float)) else "" - - is_day = int(cur.get("is_day", 1) or 1) - code = int(cur.get("weather_code", -1) or -1) + is_day_val = cur.get("is_day") + is_day_int = coerce_int(is_day_val) + is_day = is_day_int if is_day_int is not None else 1 + weather_code_val = cur.get("weather_code") + code_int = coerce_int(weather_code_val) + code = code_int if code_int is not None else -1 icon = wmo_to_icon(code, is_day) status = wmo_to_status(code) - # min/max today (index 0) - tmin_val = safe_get(daily, "temperature_2m_min", 0) - tmax_val = safe_get(daily, "temperature_2m_max", 0) - dtemp_unit = daily_units.get("temperature_2m_min", temp_unit) - tmin_str = f"{int(round(tmin_val))}{dtemp_unit}" if isinstance(tmin_val, (int, float)) else "" - tmax_str = f"{int(round(tmax_val))}{dtemp_unit}" if isinstance(tmax_val, (int, float)) else "" + tmin_val = coerce_float(safe_get(daily, "temperature_2m_min", 0)) + tmax_val = coerce_float(safe_get(daily, "temperature_2m_max", 0)) + dtemp_unit = cast(str, daily_units.get("temperature_2m_min", temp_unit)) + tmin_str = f"{int(round(tmin_val))}{dtemp_unit}" if tmin_val is not None else "" + tmax_str = f"{int(round(tmax_val))}{dtemp_unit}" if tmax_val is not None else "" min_max = f"๏ {tmin_str}\t\t๏ {tmax_str}" if tmin_str and tmax_str else "" - wind_val = cur.get("wind_speed_10m") - wind_unit = cur_units.get("wind_speed_10m", "") - wind_text = f"๎พ {int(round(wind_val))}{wind_unit}" if isinstance(wind_val, (int, float)) else "" + return temp_str, feels_str, is_day, code, icon, status, min_max - hum_val = cur.get("relative_humidity_2m") - humidity_text = f"๎ณ {int(hum_val)}%" if isinstance(hum_val, (int, float)) else "" - vis_val = cur.get("visibility") - visibility_text = f"๏ฎ {format_visibility(vis_val)}" if isinstance(vis_val, (int, float)) else "" +def build_weather_details(cur: JSONDict, cur_units: JSONDict) -> Tuple[str, str, str]: + wind_val_raw = cur.get("wind_speed_10m") + wind_val = coerce_float(wind_val_raw) + wind_unit = cast(str, cur_units.get("wind_speed_10m", "")) + if wind_val is None: + log_debug(f"Unexpected type for wind_speed_10m: {type(wind_val_raw)}") + wind_text = f" {int(round(wind_val))}{wind_unit}" if wind_val is not None else "" - aqi_val = safe_get(aqi or {}, "current", "european_aqi") - aqi_text = f"AQI {int(aqi_val)}" if isinstance(aqi_val, (int, float)) else "AQI N/A" + hum_val_raw = cur.get("relative_humidity_2m") + hum_val = coerce_float(hum_val_raw) + if hum_val is None: + log_debug(f"Unexpected type for relative_humidity_2m: {type(hum_val_raw)}") + humidity_text = f" {int(hum_val)}%" if hum_val is not None else "" - hourly_precip = build_hourly_precip(forecast) - prediction = f"\n\n{hourly_precip}" if hourly_precip else "" + vis_val_raw = cur.get("visibility") + vis_val = coerce_float(vis_val_raw) + if vis_val is None: + log_debug(f"Unexpected type for visibility: {type(vis_val_raw)}") + visibility_text = f" {format_visibility(vis_val)}" if vis_val is not None else "" - # Build place string (priority: MANUAL_PLACE > ENV_PLACE > reverse geocode > lat,lon) - place_str = (MANUAL_PLACE or ENV_PLACE or place or f"{lat:.3f}, {lon:.3f}") - location_text = f"{LOC_ICON} {place_str}" + return wind_text, humidity_text, visibility_text + + +def build_aqi_info(aqi: Optional[Dict[str, Any]]) -> str: + aqi_dict = ensure_dict(aqi) + aqi_val_raw = safe_get(aqi_dict, "current", "european_aqi") + aqi_val = coerce_float(aqi_val_raw) + if aqi_val is None: + log_debug(f"Unexpected type for european_aqi: {type(aqi_val_raw)}") + return f"AQI {int(aqi_val)}" if aqi_val is not None else "AQI N/A" + + +def build_place_str(lat: float, lon: float, place: Optional[str]) -> str: + effective_place = MANUAL_PLACE or ENV_PLACE or place + if effective_place: + return f"{effective_place} ({lat:.3f}, {lon:.3f})" + return f"{lat:.3f}, {lon:.3f}" + + + + +class TooltipParams(NamedTuple): + temp_str: str + icon: str + status: str + location_text: str + feels_str: str + min_max: str + wind_text: str + humidity_text: str + visibility_text: str + aqi_text: str + hourly_precip: str - # Build tooltip (markup or plain) + +def build_tooltip_markup(params: TooltipParams) -> str: + return str.format( + "\t\t{}\t\t\n{}\n{}\n{}\n{}\n\n{}\n{}\n{}{}", + f'<span size="xx-large">{esc(params.temp_str)}</span>', + f"<big> {params.icon}</big>", + f"<b>{esc(params.status)}</b>", + esc(params.location_text), + f"<small>{esc(params.feels_str)}</small>" if params.feels_str else "", + f"<b>{esc(params.min_max)}</b>" if params.min_max else "", + f"{esc(params.wind_text)}\t{esc(params.humidity_text)}", + f"{esc(params.visibility_text)}\t{esc(params.aqi_text)}", + f"<i> {esc(params.hourly_precip)}</i>" if params.hourly_precip else "", + ) + + +def build_tooltip_plain(params: TooltipParams) -> str: + lines = [ + f"{params.icon} {params.temp_str}", + params.status, + params.location_text, + ] + if params.feels_str: + lines.append(params.feels_str) + if params.min_max: + lines.append(params.min_max) + combined_wind = f"{params.wind_text} {params.humidity_text}".strip() + if combined_wind: + lines.append(combined_wind) + combined_visibility = f"{params.visibility_text} {params.aqi_text}".strip() + if combined_visibility: + lines.append(combined_visibility) + if params.hourly_precip: + lines.append(params.hourly_precip) + return "\n".join([ln for ln in lines if ln]) + + +def build_tooltip_text(params: TooltipParams) -> str: if TOOLTIP_MARKUP: - # Escape dynamic text to avoid breaking Pango markup - tooltip_text = str.format( - "\t\t{}\t\t\n{}\n{}\n{}\n{}\n\n{}\n{}\n{}{}", - f'<span size="xx-large">{esc(temp_str)}</span>', - f"<big> {icon}</big>", - f"<b>{esc(status)}</b>", - esc(location_text), - f"<small>{esc(feels_str)}</small>" if feels_str else "", - f"<b>{esc(min_max)}</b>" if min_max else "", - f"{esc(wind_text)}\t{esc(humidity_text)}", - f"{esc(visibility_text)}\t{esc(aqi_text)}", - f"<i> {esc(prediction)}</i>" if prediction else "", - ) + return build_tooltip_markup(params) else: - lines = [ - f"{icon} {temp_str}", - status, - location_text, - ] - if feels_str: - lines.append(feels_str) - if min_max: - lines.append(min_max) - lines.append(f"{wind_text} {humidity_text}".strip()) - lines.append(f"{visibility_text} {aqi_text}".strip()) - if prediction: - lines.append(hourly_precip) - tooltip_text = "\n".join([ln for ln in lines if ln]) + return build_tooltip_plain(params) + + +def gather_weather_data(forecast: Optional[Dict[str, Any]], aqi: Optional[Dict[str, Any]]) -> WeatherData: + forecast_dict = ensure_dict(forecast) + cur = ensure_dict(forecast_dict.get("current")) + cur_units = ensure_dict(forecast_dict.get("current_units")) + daily = ensure_dict(forecast_dict.get("daily")) + daily_units = ensure_dict(forecast_dict.get("daily_units")) + + temp_str, feels_str, is_day, code, icon, status, min_max = build_weather_strings(cur, cur_units, daily, daily_units, cast(str, cur_units.get("temperature_2m", ""))) + wind_text, humidity_text, visibility_text = build_weather_details(cur, cur_units) + aqi_text = build_aqi_info(aqi) + hourly_precip = build_hourly_precip(forecast_dict) + + return WeatherData( + temp_str=temp_str, + feels_str=feels_str, + icon=icon, + status=status, + min_max=min_max, + wind_text=wind_text, + humidity_text=humidity_text, + visibility_text=visibility_text, + aqi_text=aqi_text, + hourly_precip=hourly_precip, + is_day=is_day, + code=code, + ) + + +def build_output(loc: Location, forecast: Optional[Dict[str, Any]], aqi: Optional[Dict[str, Any]]) -> Tuple[Dict[str, str], str]: + data = gather_weather_data(forecast, aqi) + + place_str = build_place_str(loc.lat, loc.lon, loc.place) + location_text = f"{LOC_ICON} {place_str}" + + tooltip_text = build_tooltip_text( + TooltipParams( + data.temp_str, data.icon, data.status, location_text, data.feels_str, data.min_max, + data.wind_text, data.humidity_text, data.visibility_text, data.aqi_text, data.hourly_precip + ) + ) - out_data = { - "text": f"{icon} {temp_str}", - "alt": status, + out_data: Dict[str, Any] = { + "text": f"{data.icon} {data.temp_str}", + "alt": data.status, "tooltip": tooltip_text, - "class": f"wmo-{code} {'day' if is_day else 'night'}", + "class": f"wmo-{data.code} {'day' if data.is_day else 'night'}", } simple_weather = ( - f"{icon} {status}\n" - + f"๏ {temp_str} ({feels_str})\n" - + (f"{wind_text} \n" if wind_text else "") - + (f"{humidity_text} \n" if humidity_text else "") - + f"{visibility_text} {aqi_text}\n" + f"{place_str}\n" + f"{data.icon} {data.status}\n" + + f"๏ {data.temp_str} ({data.feels_str})\n" + + (f"๎พ {data.wind_text} \n" if data.wind_text else "") + + (f"๎ณ {data.humidity_text} \n" if data.humidity_text else "") + + f"๏ฎ {data.visibility_text} {data.aqi_text}\n" ) return out_data, simple_weather -def main() -> None: - lat, lon = get_coords() - - # Try cache first +def try_cached_weather(lat: float, lon: float) -> Optional[Tuple[Dict[str, str], str]]: cached = read_api_cache() - if cached and isinstance(cached, dict): - forecast = cached.get("forecast") - aqi = cached.get("aqi") - cached_place = cached.get("place") if isinstance(cached.get("place"), str) else None - place_effective = MANUAL_PLACE or ENV_PLACE or cached_place + if cached: + forecast = cast(Optional[Dict[str, Any]], cached.get("forecast")) + aqi = cast(Optional[Dict[str, Any]], cached.get("aqi")) + place_val = cached.get("place") + cached_place = place_val if isinstance(place_val, str) else None + # Ensure the cached forecast corresponds to the requested lat/lon + fc = ensure_dict(cached.get("forecast")) + c_lat = coerce_float(safe_get(fc, "latitude")) + c_lon = coerce_float(safe_get(fc, "longitude")) + if c_lat is not None and c_lon is not None: + if abs(c_lat - lat) > 0.1 or abs(c_lon - lon) > 0.1: + return None # force fresh fetch for new location try: - out, simple = build_output(lat, lon, forecast, aqi, place_effective) - print(json.dumps(out, ensure_ascii=False)) - write_simple_text_cache(simple) - return + return build_output(Location(lat, lon, cached_place), forecast, aqi) except Exception as e: print(f"Cached data build failed, refetching: {e}", file=sys.stderr) + return None - # Fetch fresh + +def fetch_fresh_weather(lat: float, lon: float) -> Optional[Tuple[Dict[str, str], str]]: try: forecast = fetch_open_meteo(lat, lon) aqi = fetch_aqi(lat, lon) - # Use manual/env place if provided; otherwise reverse geocode - place_effective = MANUAL_PLACE or ENV_PLACE or fetch_place(lat, lon) - write_api_cache({"forecast": forecast, "aqi": aqi, "place": place_effective}) - out, simple = build_output(lat, lon, forecast, aqi, place_effective) - print(json.dumps(out, ensure_ascii=False)) - write_simple_text_cache(simple) + # If MANUAL_PLACE is set, don't reverse geocode - use the manual place instead + place = MANUAL_PLACE if MANUAL_PLACE else fetch_place(lat, lon) + write_api_cache({"forecast": forecast, "aqi": aqi, "place": place}) + return build_output(Location(lat, lon, place), forecast, aqi) except Exception as e: print(f"Open-Meteo fetch failed: {e}", file=sys.stderr) - # Last resort: try stale cache without TTL - try: - if os.path.exists(API_CACHE_PATH): - with open(API_CACHE_PATH, "r", encoding="utf-8") as f: - stale = json.load(f) - out, simple = build_output(lat, lon, stale.get("forecast", {}), stale.get("aqi"), stale.get("place") if isinstance(stale.get("place"), str) else None) - print(json.dumps(out, ensure_ascii=False)) - write_simple_text_cache(simple) - return - except Exception as e2: - print(f"Failed to use stale cache: {e2}", file=sys.stderr) - # Fallback minimal output - fallback = { - "text": f"{WEATHER_ICONS['default']} N/A", - "alt": "Unavailable", - "tooltip": "Weather unavailable", - "class": "unavailable", - } - print(json.dumps(fallback, ensure_ascii=False)) + return None + + +def try_stale_weather(lat: float, lon: float) -> Optional[Tuple[Dict[str, str], str]]: + try: + if API_CACHE_PATH.exists(): + with API_CACHE_PATH.open("r", encoding="utf-8") as f: + stale = json.load(f) + stale_dict = ensure_dict(stale) + place_val = stale_dict.get("place") + place = place_val if isinstance(place_val, str) else None + forecast = cast(Optional[Dict[str, Any]], stale_dict.get("forecast")) + aqi = cast(Optional[Dict[str, Any]], stale_dict.get("aqi")) + return build_output(Location(lat, lon, place), forecast, aqi) + except Exception as e2: + print(f"Failed to use stale cache: {e2}", file=sys.stderr) + return None + + +def main() -> None: + lat, lon = get_coords() + + # Try cache first + result = try_cached_weather(lat, lon) + if result: + out, simple = result + print(json.dumps(out, ensure_ascii=False)) + write_simple_text_cache(simple) + return + + # Fetch fresh + result = fetch_fresh_weather(lat, lon) + if result: + out, simple = result + print(json.dumps(out, ensure_ascii=False)) + write_simple_text_cache(simple) + return + + # Last resort: try stale cache + result = try_stale_weather(lat, lon) + if result: + out, simple = result + print(json.dumps(out, ensure_ascii=False)) + write_simple_text_cache(simple) + return + + # Fallback minimal output + fallback = { + "text": f"{WEATHER_ICONS['default']} N/A", + "alt": "Unavailable", + "tooltip": "Weather unavailable", + "class": "unavailable", + } + print(json.dumps(fallback, ensure_ascii=False)) + + +def test_coerce_functions(): + """Manual testing for coerce functions.""" + # Test coerce_int + assert coerce_int(5) == 5 + assert coerce_int(5.5) == 5 + assert coerce_int("5") == 5 + assert coerce_int("5.7") == 5 + assert coerce_int("abc") is None + assert coerce_int(None) is None + + # Test coerce_float + assert coerce_float(5.5) == 5.5 + assert coerce_float(5) == 5.0 + assert coerce_float("5.5") == 5.5 + assert coerce_float("abc") is None + assert coerce_float(None) is None + + # Test coerce_number + assert coerce_number(5) == 5 + assert coerce_number(5.5) == 5.5 + assert coerce_number("5") == 5 + assert coerce_number("5.5") == 5.5 + assert coerce_number("abc") is None + + print("All coerce function tests passed.", file=sys.stderr) if __name__ == "__main__": - main() + if len(sys.argv) > 1 and sys.argv[1] == "--test": + test_coerce_functions() + else: + main() diff --git a/config/hypr/UserScripts/Weather.sh b/config/hypr/UserScripts/Weather.sh index 9bdaff4a..ac9abc13 100755 --- a/config/hypr/UserScripts/Weather.sh +++ b/config/hypr/UserScripts/Weather.sh @@ -1,18 +1,42 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## # weather info from wttr. https://github.com/chubin/wttr.in # Remember to add city -city= +city="" + + +# if city is blank, use https://ipapi.co/json to get location from IP +if [ -z "$city" ]; then + city=$(curl -fsS https://ipapi.co/json | grep city | cut -f4 -d'"') +fi + + +# URL-encode city for safe use in URLs +encoded_city="$city" +if command -v python3 >/dev/null 2>&1; then + encoded_city=$(python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1]))' "$city") +elif command -v jq >/dev/null 2>&1; then + encoded_city=$(printf '%s' "$city" | jq -sRr @uri) +else + # Minimal fallback: encode a few common special characters + encoded_city=$(printf '%s' "$city" | sed -e 's/ /%20/g' -e 's/&/%26/g' -e 's/?/%3F/g' -e 's/#/%23/g') +fi + + cachedir="$HOME/.cache/rbn" -cachefile=${0##*/}-$1 +# Include city and arg in cache key so changing city invalidates old cache +cache_key="${city}_${1}" +# Sanitize cache key to avoid problematic characters in filename +safe_key=$(printf '%s' "$cache_key" | tr -c '[:alnum:]_-' '_') +cachefile=${0##*/}-$safe_key -if [ ! -d $cachedir ]; then - mkdir -p $cachedir +if [ ! -d "$cachedir" ]; then + mkdir -p "$cachedir" fi -if [ ! -f $cachedir/$cachefile ]; then - touch $cachedir/$cachefile +if [ ! -f "$cachedir/$cachefile" ]; then + touch "$cachedir/$cachefile" fi # Save current IFS @@ -20,25 +44,88 @@ SAVEIFS=$IFS # Change IFS to new line. IFS=$'\n' -cacheage=$(($(date +%s) - $(stat -c '%Y' "$cachedir/$cachefile"))) -if [ $cacheage -gt 1740 ] || [ ! -s $cachedir/$cachefile ]; then - data=($(curl -s https://en.wttr.in/"$city"$1\?0qnT 2>&1)) - echo ${data[0]} | cut -f1 -d, > $cachedir/$cachefile - echo ${data[1]} | sed -E 's/^.{15}//' >> $cachedir/$cachefile - echo ${data[2]} | sed -E 's/^.{15}//' >> $cachedir/$cachefile +file="$cachedir/$cachefile" +# Portable file mtime retrieval (GNU/BSD): +# - GNU: stat -c %Y <file> +# - BSD/macOS: stat -f %m <file> +mtime=$(stat -c %Y "$file" 2>/dev/null || stat -f %m "$file" 2>/dev/null || echo 0) +now=$(date +%s) +cacheage=$(( now - mtime )) +if [ $cacheage -gt 1740 ] || [ ! -s "$cachedir/$cachefile" ]; then + # Prefer structured format for reliable parsing (3 lines: location, condition, temperature) + mapfile -t sdata < <(curl -fsS "https://wttr.in/${encoded_city}?format=%25l%0A%25C%0A%25t&lang=en" 2>/dev/null || true) + if [ ${#sdata[@]} -ge 3 ]; then + printf "%s\n" "${sdata[0]}" > "$cachedir/$cachefile" + printf "%s\n" "${sdata[1]}" >> "$cachedir/$cachefile" + printf "%s\n" "${sdata[2]}" >> "$cachedir/$cachefile" + else + # Try fetching each field separately if combined format is flaky + loc=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25l&lang=en" 2>/dev/null || true) + cond_only=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25C&lang=en" 2>/dev/null || true) + temp_only=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25t" 2>/dev/null || true) + if [ -n "$loc" ] && [ -n "$cond_only" ] && [ -n "$temp_only" ]; then + printf "%s\n" "$loc" > "$cachedir/$cachefile" + printf "%s\n" "$cond_only" >> "$cachedir/$cachefile" + printf "%s\n" "$temp_only" >> "$cachedir/$cachefile" + else + # Fallback: try ASCII output and extract best-effort fields + url="https://en.wttr.in/${encoded_city}?1" + mapfile -t data < <(curl -fsS "$url" 2>/dev/null || true) + if [ ${#data[@]} -ge 3 ] && ! echo "${data[0]}" | grep -qi 'not found\|unknown location'; then + loc=$(echo "${data[0]}" | sed -E 's/^.*: *//') + # Attempt to pull condition and temperature hints from nearby lines + cond=$(echo "${data[2]}" | sed -E 's/^.{0,15}//; s/^\s+//') + temp=$(printf "%s\n" "${data[@]}" | grep -Eo '\+?-?[0-9]+(\([^)]+\))? ?ยฐ?[CF]' | head -n1) + # Only write if we have at least location and something else meaningful + if [ -n "$loc" ] && { [ -n "$cond" ] || [ -n "$temp" ]; }; then + printf "%s\n" "$loc" > "$cachedir/$cachefile" + printf "%s\n" "${cond:-Unknown}" >> "$cachedir/$cachefile" + printf "%s\n" "${temp:-N/A}" >> "$cachedir/$cachefile" + fi + fi + fi + fi fi -weather=($(cat $cachedir/$cachefile)) +# Read cache robustly (line-wise) +mapfile -t weather < "$cachedir/$cachefile" + +# If cache is still empty or invalid, emit a single error JSON and exit to avoid double-prints +if [ ${#weather[@]} -lt 3 ] || ! echo "${weather[2]}" | grep -qE '[-+0-9].*ยฐ'; then + # Last-chance: try live structured fetch and populate cache and runtime weather + mapfile -t sdata < <(curl -fsS "https://wttr.in/${encoded_city}?format=%25l%0A%25C%0A%25t&lang=en" 2>/dev/null || true) + if [ ${#sdata[@]} -ge 3 ]; then + weather=("${sdata[@]}") + printf "%s\n" "${sdata[0]}" > "$cachedir/$cachefile" + printf "%s\n" "${sdata[1]}" >> "$cachedir/$cachefile" + printf "%s\n" "${sdata[2]}" >> "$cachedir/$cachefile" + else + loc=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25l&lang=en" 2>/dev/null || true) + cond_only=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25C&lang=en" 2>/dev/null || true) + temp_only=$(curl -fsS "https://wttr.in/${encoded_city}?format=%25t" 2>/dev/null || true) + if [ -n "$loc" ] && [ -n "$cond_only" ] && [ -n "$temp_only" ]; then + weather=("$loc" "$cond_only" "$temp_only") + printf "%s\n" "$loc" > "$cachedir/$cachefile" + printf "%s\n" "$cond_only" >> "$cachedir/$cachefile" + printf "%s\n" "$temp_only" >> "$cachedir/$cachefile" + else + echo -e "{\"text\":\"\uf06a\", \"alt\":\"\", \"tooltip\":\": \"}" + exit 1 + fi + fi +fi # Restore IFSClear IFS=$SAVEIFS -temperature=$(echo ${weather[2]} | sed -E 's/([[:digit:]]+)\.\./\1 to /g') +temperature=$(echo "${weather[2]}" | sed -E 's/([[:digit:]]+)\.\./\1 to /g') #echo ${weather[1]##*,} # https://fontawesome.com/icons?s=solid&c=weather -case $(echo ${weather[1]##*,} | tr '[:upper:]' '[:lower:]') in +# Normalize condition string for matching +cond_key=$(echo "${weather[1]##*,}" | tr '[:upper:]' '[:lower:]' | sed -E 's/^\s+//; s/\s+$//') +case "$cond_key" in "clear" | "sunny") condition="๎" ;; @@ -54,7 +141,7 @@ case $(echo ${weather[1]##*,} | tr '[:upper:]' '[:lower:]') in "fog" | "freezing fog") condition="๎" ;; -"patchy rain possible" | "patchy light drizzle" | "light drizzle" | "patchy light rain" | "light rain" | "light rain shower" | "mist" | "rain") +"patchy rain possible" | "patchy light drizzle" | "light drizzle" | "patchy light rain" | "light rain" | "light rain shower" | "mist" | "rain" | "patchy rain nearby") condition="๓ฐผณ" ;; "moderate rain at times" | "moderate rain" | "heavy rain at times" | "heavy rain" | "moderate or heavy rain shower" | "torrential rain shower" | "rain shower") @@ -74,14 +161,49 @@ case $(echo ${weather[1]##*,} | tr '[:upper:]' '[:lower:]') in ;; *) condition="๏ช" - echo -e "{\"text\":\""$condition"\", \"alt\":\""${weather[0]}"\", \"tooltip\":\""${weather[0]}: $temperature ${weather[1]}"\"}" ;; esac +# If still unknown, try substring heuristics to pick a reasonable icon +if [ "$condition" = "๏ช" ]; then + if echo "$cond_key" | grep -q "rain\|drizzle\|shower"; then + condition="๓ฐผณ" + elif echo "$cond_key" | grep -q "heavy rain\|torrential"; then + condition="๎น" + elif echo "$cond_key" | grep -q "snow"; then + condition="๓ฐฟ" + elif echo "$cond_key" | grep -q "sleet\|freezing\|ice"; then + condition="๓ฐผด" + elif echo "$cond_key" | grep -q "thunder"; then + condition="๎" + elif echo "$cond_key" | grep -q "overcast"; then + condition="๎" + elif echo "$cond_key" | grep -q "cloud"; then + condition="๎" + elif echo "$cond_key" | grep -q "sunny\|clear"; then + condition="๎" + fi +fi + #echo $temp $condition -echo -e "{\"text\":\""$temperature $condition"\", \"alt\":\""${weather[0]}"\", \"tooltip\":\""${weather[0]}: $temperature ${weather[1]}"\"}" +# Ensure temperature has a value; if empty, keep whatever is in weather[2] or N/A +if [ -z "$temperature" ]; then + temperature="${weather[2]:-N/A}" +fi + +cond_disp=$(echo "${weather[1]}" | sed -E 's/^\s+//; s/\s+$//') + +# Escape strings for safe JSON embedding (escape backslashes and double quotes) +json_escape() { + printf '%s' "$1" | sed -e 's/\\/\\\\/g' -e 's/\"/\\\"/g' +} + +text_json=$(json_escape "$temperature $condition") +alt_json=$(json_escape "${weather[0]}") +tooltip_json=$(json_escape "${weather[0]}: $temperature $cond_disp") -cached_weather="๏ $temperature \n$condition ${weather[1]}" +printf '{"text":"%s", "alt":"%s", "tooltip":"%s"}\n' "$text_json" "$alt_json" "$tooltip_json" -echo -e $cached_weather > "$HOME/.cache/.weather_cache"
\ No newline at end of file +# Write a two-line cache with an actual newline between lines +printf '๏ %s \n%s %s\n' "$temperature" "$condition" "${weather[1]}" > "$HOME/.cache/.weather_cache"
\ No newline at end of file diff --git a/config/hypr/UserScripts/WeatherWrap.sh b/config/hypr/UserScripts/WeatherWrap.sh new file mode 100755 index 00000000..10c125dc --- /dev/null +++ b/config/hypr/UserScripts/WeatherWrap.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## +# Weather entrypoint: prefer Python (OpenโMeteo), fallback to legacy Bash (wttr.in) + +SCRIPT_DIR="$(dirname "$0")" +PY_SCRIPT="$SCRIPT_DIR/Weather.py" +BASH_FALLBACK="$SCRIPT_DIR/Weather.sh" + +run_fallback() { + if [ -f "$BASH_FALLBACK" ]; then + # Invoke via bash to avoid requiring +x and ensure consistent shell + bash "$BASH_FALLBACK" "$@" + return $? + else + echo "Weather fallback not found: $BASH_FALLBACK" >&2 + return 127 + fi +} + +if command -v python3 >/dev/null 2>&1; then + python3 "$PY_SCRIPT" "$@" + exit_code=$? + if [ "$exit_code" -eq 0 ]; then + exit 0 + fi + echo "Weather.py failed with code $exit_code โ falling back to Weather.sh" >&2 + run_fallback "$@" + exit $? +else + echo "python3 not found in PATH โ falling back to Weather.sh" >&2 + run_fallback "$@" + exit $? +fi
\ No newline at end of file diff --git a/config/hypr/UserScripts/ZshChangeTheme.sh b/config/hypr/UserScripts/ZshChangeTheme.sh index cffaf5cb..690f0f13 100755 --- a/config/hypr/UserScripts/ZshChangeTheme.sh +++ b/config/hypr/UserScripts/ZshChangeTheme.sh @@ -1,4 +1,4 @@ -#!/bin/bash +#!/usr/bin/env bash # /* ---- ๐ซ https://github.com/JaKooLit ๐ซ ---- */ ## # Script for Oh my ZSH theme ( CTRL SHIFT O) |
