203 lines
7.2 KiB
Python
Executable File
203 lines
7.2 KiB
Python
Executable File
#!/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: <https://github.com/MatthewFlamm/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
|
||
#
|
||
# N.B.: using only one provider is risky in case of API change (or simple outage).
|
||
# - see, e.g.: <https://gitlab.gnome.org/GNOME/libgweather/-/issues/236>
|
||
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: <https://aviationweather.gov/metar>'
|
||
)
|
||
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()
|