#!/usr/bin/env python3 # to get GWeather logging, run with # `G_MESSAGES_DEBUG=GWeather` # queries weather from public (gov) sources and renders it to stdout. # primary consumer is conky. # very limited; libgweather is a little iffy and doesn't expose everything the raw APIs do # e.g. no precise precipitation predictions. # # future work: # - consider using python-native pynws: # - it's well-factored, exposing a nice interface but letting me dig as deep as i want wherever that's incomplete. # - render a graph with rain + temperature forecast # - conky supports graphs with [execgraph](https://conky.cc/variables#execgraph) import argparse import code import gi import logging import time gi.require_version('GWeather', '4.0') from gi.repository import GLib from gi.repository import GWeather logger = logging.getLogger(__name__) class WeatherSource: ''' this class abstracts operations which would query a remote weather server ''' def __init__(self): self.info = GWeather.Info() self.info.set_application_id('org.uninsane.sane-weather') self.info.set_contact_info('contact@uninsane.org') self.info.set_enabled_providers( # defaults to METAR | IWIN. # options are: # - IWIN # graphical.weather.gov; provides daily min/max temp, precipitation # - METAR # aviationweather.gov; provides current time, wind, visibility, conditions, clouds, temperature, pressure # - MET_NO # - NWS # api.weather.gov; provides hourly temperature, dewpoint, humidity, sky cover, wind, precipitation, snow; daily min/max temp, # - OWM # METAR, if you only want immediate conditions GWeather.Provider.METAR # METAR + NWS, if you want a forecast # GWeather.Provider.METAR | GWeather.Provider.NWS ) self.world = GWeather.Location.get_world() def query_loc(self, loc: GWeather.Location) -> None: ''' query the weather for some location, asynchronously. after calling, poll the `try_...` methods to check for results. ''' logger.debug(f"querying: {loc.get_coords()}") self.info.set_location(loc) self.info.update() def try_get_celcius(self) -> float | None: valid, temp = self.info.get_value_temp(GWeather.TemperatureUnit.CENTIGRADE) logger.debug(f"try_get_celcius: valid={valid}, temp={temp}") if not valid: temp = None return temp # potentially interesting methods on GWeather.Info: # - get_conditions # returns '-' # - get_forecast_list # forecast as a list of GWeather.Info instances (daily if IWIN; hourly if NWS) # - get_sky # like 'Clear sky' # - get_sunrise, get_sunset # like '13∶10', '03∶04' (utc time) # - get_symbolic_icon_name # like 'weather-clear-night-symbolic' # - get_temp_min, get_temp_max # returns '-' # - get_temp_summary() # same as get_temp() # - get_update() # like 'Thu, Aug 24 / 13∶00' # - get_wind() # like 'North / 13.0 km/h' # - get_visibility() # like '16093m' # - get_weather_summary() # like 'Seattle-Tacoma International Airport: Clear sky' class TopLevel: """ this class acts as the "event loop" which glib apps expect. caller sets up a "work queue" of everything they want to do, then calls `run`. glib calls `poll` in a loop, and each time we try to work through another item in the work_queue. when the work_queue is empty, exit glib's main loop & return to the caller (from `run`). """ def __init__(self): self._loop = GLib.MainLoop() self.source = WeatherSource() self.work_queue = [] def enqueue(self, op) -> None: self.work_queue.append(op) def run(self) -> None: self.enqueue(ExitOp()) GLib.idle_add(self.poll) self._loop.run() def poll(self) -> bool: work = self.work_queue[0] if isinstance(work, QueryOp): del self.work_queue[0] self.source.query_loc(work.loc) elif isinstance(work, PrintTempOp): temp = self.source.try_get_celcius() if temp is not None: del self.work_queue[0] print(f"{int(temp)}°C") elif isinstance(work, DiagnosticsOp): del self.work_queue[0] # GWeather does transparent caching so that we don't usually hit the web last_update = self.source.info.get_update() logger.debug(f"last update: {last_update}") elif isinstance(work, ExitOp): logger.debug("quitting GLib MainLoop") self.source.info.store_cache() self._loop.quit() elif isinstance(work, IdleOp): del self.work_queue[0] logger.debug("micro sleep") time.sleep(0.1) else: assert False, f"unknown work: {work}" # micro sleep so we don't peg CPU # TODO: i'm sure there's a better way than all of this time.sleep(0.05) # re-queue this idle fn return True # operations: # think of these as public methods on the `TopLevel` class, # except abstracted as values for the sake of glib's event loop. class QueryOp: def __init__(self, loc: GWeather.Location): self.loc = loc class PrintTempOp: pass class DiagnosticsOp: pass class IdleOp: pass class ExitOp: pass def main(): logging.basicConfig() parser = argparse.ArgumentParser(description="acquire weather information for user display") parser.add_argument( '--station-code', default='KSEA', help='4-letter METAR weather station code for where we want to know weather\n ' 'to find your station see here: ' ) parser.add_argument('--break-before', action='store_true', help='drop into a REPL before do anything (for debugging)') parser.add_argument('--break-after', action='store_true', help='drop into a REPL after completing the work (for debugging)') parser.add_argument('--verbose', action='store_true', help='enable verbose logging') args = parser.parse_args() if args.verbose: logger.setLevel(logging.DEBUG) GLib.log_set_debug_enabled(True) toplevel = TopLevel() here = GWeather.Location.find_by_station_code(toplevel.source.world, args.station_code) if args.break_before: code.interact(local=dict(**globals(), **locals())) toplevel.enqueue(QueryOp(here)) toplevel.enqueue(PrintTempOp()) toplevel.enqueue(DiagnosticsOp()) # for _ in range(300): # for debugging... # toplevel.enqueue(IdleOp()) toplevel.run() if args.break_after: code.interact(local=dict(**globals(), **locals())) if __name__ == '__main__': main()