aboutsummaryrefslogtreecommitdiffstats
path: root/config/hypr/UserScripts
diff options
context:
space:
mode:
Diffstat (limited to 'config/hypr/UserScripts')
-rwxr-xr-xconfig/hypr/UserScripts/RainbowBorders.sh2
-rwxr-xr-xconfig/hypr/UserScripts/RofiBeats.sh203
-rwxr-xr-xconfig/hypr/UserScripts/RofiCalc.sh2
-rwxr-xr-xconfig/hypr/UserScripts/Tak0-Autodispatch.sh2
-rwxr-xr-xconfig/hypr/UserScripts/WallpaperAutoChange.sh2
-rwxr-xr-xconfig/hypr/UserScripts/WallpaperEffects.sh2
-rwxr-xr-xconfig/hypr/UserScripts/WallpaperRandom.sh2
-rwxr-xr-xconfig/hypr/UserScripts/WallpaperSelect.sh2
-rwxr-xr-xconfig/hypr/UserScripts/Weather.py767
-rwxr-xr-xconfig/hypr/UserScripts/Weather.sh164
-rwxr-xr-xconfig/hypr/UserScripts/WeatherWrap.sh33
-rwxr-xr-xconfig/hypr/UserScripts/ZshChangeTheme.sh2
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)
send patches to the email below
yukais@pinapelz.com
include the subject [PATCH repo_name]
pinapelz.com
homepage