Plaster
New
List
Login
python
default
isoraqathedh
2022.04.14 13:46:30
#!/usr/bin/python from pint import UnitRegistry from datetime import datetime import colorlover as colours import json import pathlib import os import sys # Load data units = UnitRegistry() home = "~/Documents/Junk/weather-archives/" select = ((len(sys.argv) >= 2 and sys.argv[1]) or os.getenv("BLOCK_INSTANCE") or "temperature") with open(max(pathlib.PosixPath(home).expanduser().glob("owm_*.json"), key=lambda x: x.stat().st_ctime)) as f: dat = json.load(f) # Create results dict: if select == "temperature": res = { "val": (dat["current"]["temp"] * units.kelvin).to("degC").magnitude, "fmt": "{:5.2f}°C", "null_string": "--.-- °C", "scl_t": "div", "scl_n": "RdYlBu", "reverse": True, "bounds": [7, 12, 17, 22, 27, 32], } elif select == "apparent_temperature": res = { "val": ((dat["current"]["feels_like"] * units.kelvin).to("degC").magnitude if abs(dat["current"]["feels_like"] - dat["current"]["temp"]) >= 2 else None), "fmt": "({:5.2f}°C)", "null_string": "(--.--°C)", "scl_t": "div", "scl_n": "RdYlBu", "reverse": True, "bounds": [7, 12, 17, 22, 27, 32], } elif select == "wind_speed": res = { "val": ((dat["current"]["wind_speed"] * units.metre / units.second) .to("km/h").magnitude), "fmt": "{:5.1f} km/h", "null_string": "---.-- km/h", "scl_t": "seq", "scl_n": "Greens", "reverse": True, "bounds": [0, 12, 30, 40, 62, 87, 117] } elif select == "wind_gust": res = { "val": (((dat["current"]["wind_gust"] * units.metre / units.second) .to("km/h").magnitude) if "wind_gust" in dat["current"] and abs(dat["current"]["wind_gust"] - dat["current"]["wind_speed"]) >= 5 else None), "fmt": "({:5.1f} km/h)", "null_string": "(---.-- km/h)", "scl_t": "seq", "scl_n": "Greens", "reverse": True, "bounds": [0, 12, 30, 40, 62, 87, 117] } elif select == "wind_deg": res = { "val": dat["current"]["wind_deg"], "fmt": "{:03d}°", "null_string": "---°", "scl_t": None } elif select == "pressure": res = { "val": dat["current"]["pressure"], "fmt": "{:4.1f} hPa", "null_string": "----.- hPa", "scl_t": "div", "scl_n": "RdYlBu", "reverse": True, "bounds": [995, 1000, 1005, 1010, 1015, 1020, 1025] } elif select == "visibility": res = { "val": ((dat["current"]["visibility"] * units.metre).to("km").magnitude if dat["current"]["visibility"] < 10000 else None), "fmt": "{:4.2f} km", "null_string": "-.-- km", "scl_t": "seq", "scl_n": "Reds", "reverse": True, "bounds": [0.5, 3, 5, 10] } elif select == "humidity": res = { "val": dat["current"]["humidity"], "fmt": "{:3.0f}%", "null_string": "---%", "scl_t": "div", "scl_n": "RdYlBu", "reverse": False, "bounds": [30, 50, 70, 80, 90, 95] } elif select == "UV": current = dat["current"] res = { "val": (current["uvi"] if current["uvi"] >= 2 and current["sunrise"] < current["dt"] < current["sunset"] else None), "fmt": "U{:2d}", "null_string": "U -", "scl_t": "seq", "scl_n": "Purples", "bounds": [0, 1, 5, 7, 9] } elif select == "now": res = { "val": datetime.fromtimestamp(dat["current"]["dt"]).strftime("🕛 %d%H%MJ"), "fmt": "{}", "scl_t": None } elif select == "sunrise": res = { "val": datetime.fromtimestamp(dat["current"]["sunrise"]).strftime("☀↑ %H:%M:%S"), "fmt": "{}", "scl_t": None } elif select == "sunset": res = { "val": datetime.fromtimestamp(dat["current"]["sunset"]).strftime("☀↓ %H:%M:%S"), "fmt": "{}", "scl_t": None } elif select == "rain": # Actual rain and probability to be figured out later. cloud_level = dat["current"]["clouds"] if cloud_level > 0: res = { "val": cloud_level, "fmt": "☁ {:2d}%", "scl_t": "seq", "scl_n": "Blues", "reverse": True, "bounds": [0, 30, 70, 100] } else: res = { "val": 0, "fmt": "☁⃠", "scl_t": None } elif select == "status": res = { "val": dat["current"]["weather"][0]["main"], "fmt": "{}", "scl_t": None, } elif select == "hilo": time_horizon = 12 if 5 < datetime.fromtimestamp(dat["current"]["dt"]).hour < 17: maxloc, maxtemp = max(enumerate(dat["hourly"][:time_horizon]), key=lambda x: x[1]["temp"]) mintemp = min(dat["hourly"][maxloc : maxloc + time_horizon], key=lambda x: x["temp"]) res = { "val": [(maxtemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(maxtemp["dt"]).hour, (mintemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(mintemp["dt"]).hour,], "fmt": ('<span background="#801A00">↑ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span background="#005266">↓ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "scl_t": None } else: minloc, mintemp = min(enumerate(dat["hourly"][:time_horizon]), key=lambda x: x[1]["temp"]) maxtemp = max(dat["hourly"][minloc : minloc + time_horizon], key=lambda x: x["temp"]) res = { "val": [(mintemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(mintemp["dt"]).hour, (maxtemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(maxtemp["dt"]).hour], "fmt": ('<span background="#005266">↓ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span background="#801A00">↑ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "scl_t": None } # Format results dict: if res["val"] is not None: print(res["fmt"].format(res["val"])) print(res["fmt"].format(res["val"])) if res["scl_t"]: selected_scale = colours.to_numeric(colours.scales[ str(len(res["bounds"]))][ res["scl_t"]][ res["scl_n"]][ ::(-1 if res["reverse"] else 1)]) ordinary = res["bounds"][0] <= res["val"] < res["bounds"][-1] if ordinary: red, green, blue = tuple( map(int, [c for (t, c) in zip(res["bounds"], selected_scale) if res["val"] < t][-1])) print(f"#{red:02X}{green:02X}{blue:02X}") sys.exit(0) else: sys.exit(33) else: sys.exit(0) else: print(res["null_string"]) print(res["null_string"]) print("#999999") sys.exit(0)
Raw
Annotate
Repaste
Annotations
python
default
isoraqathedh
2022.04.15 06:00:06
#!/usr/bin/python from pint import UnitRegistry from datetime import datetime import colorlover as colours import json import pathlib import os import sys units = UnitRegistry() # Data types: class Colour_spec: scale_type = "" scale_name = "" reverse = False bounds = [] def __init__(self, s_type, s_name, bounds, rev=False): self.scale_type = s_type self.scale_name = s_name self.bounds = bounds self.reverse = rev def is_urgent(self, value): return not self.bounds[0] <= value < self.bounds[-1] def colour_of(self, value): selected_scale = colours.to_numeric( colours.scales [str(len(self.bounds))] [self.scale_type] [self.scale_name] [::(-1 if self.reverse else 1)]) if not self.is_urgent(value): red, green, blue = tuple( map(int, [c for (t, c) in zip(self.bounds, selected_scale) if value < t][-1])) return f"#{red:02X}{green:02X}{blue:02X}" else: return None # Colour specs: colour_specs = { "temperature": Colour_spec("div", "RdYlBu", [7, 12, 17, 22, 27, 32], True), "wind": Colour_spec("seq", "Greens", [0, 12, 30, 40, 62, 87, 117], True), } # Result organisation and formatting class Result: val = None fmt = "" null_string = None colours = None def __init__(self, val, fmt, null_string, colours=None): self.val = val self.fmt = fmt self.null_string = null_string self.colours = colours def value_string(self): return (self.fmt.format(self.val) if self.val is not None else self.null_string) def return_code(self): return 33 if self.colours and self.colours.is_urgent(self.val) else 0 def colour(self): if self.val is None: return "#999999" elif self.colours: return self.colours.colour_of(self.val) else: return None def print_results(self): print(self.value_string()) print(self.value_string()) c = self.colour() if c: print(c) # Processing the JSON file: def load(home): with open(max(pathlib.PosixPath(home).expanduser().glob("owm_*.json"), key=lambda x: x.stat().st_ctime)) as f: return json.load(f) def collect_owm_data(dat, select): global units if select == "temperature": res_o = Result( (dat["current"]["temp"] * units.kelvin).to("degC").magnitude, "{:5.2f}°C", "--.-- °C", colour_specs["temperature"] ) elif select == "apparent_temperature": apparent = (dat["current"]["feels_like"] * units.kelvin).to("degC").magnitude real = (dat["current"]["temp"] * units.kelvin).to("degC").magnitude res_o = Result( apparent if abs(apparent - real) > 2 else None, "({:5.2f}°C)", "(--.--°C)", colour_specs["temperature"] ) elif select == "wind_speed": res_o = Result( ((dat["current"]["wind_speed"] * units.metre / units.second) .to("km/h").magnitude), "{:5.1f} km/h", "---.-- km/h", colour_specs["wind"] ) elif select == "wind_gust": res_o = Result( (((dat["current"]["wind_gust"] * units.metre / units.second) .to("km/h").magnitude) if "wind_gust" in dat["current"] and abs(dat["current"]["wind_gust"] - dat["current"]["wind_speed"]) >= 5 else None), "({:5.1f} km/h)", "(---.-- km/h)", colour_specs["wind"] ) elif select == "wind_deg": res_o = Result( dat["current"]["wind_deg"], "{:03d}°", "---°", None ) elif select == "pressure": res_o = Result( dat["current"]["pressure"], "{:4.1f} hPa", "----.- hPa", Colour_spec("div", "RdYlBu", [995, 1000, 1005, 1010, 1015, 1020, 1025], True) ) elif select == "visibility": res_o = Result( ((dat["current"]["visibility"] * units.metre).to("km").magnitude if dat["current"]["visibility"] < 10000 else None), "{:4.2f} km", "-.-- km", Colour_spec("seq", "Reds", [0.5, 3, 5, 10], True) ) elif select == "humidity": res_o = Result( dat["current"]["humidity"], "{:3.0f}%", "---%", Colour_spec("div", "RdYlBu", [30, 50, 70, 80, 90, 95]) ) elif select == "UV": current = dat["current"] res_o = Result( (current["uvi"] if current["uvi"] >= 2 and current["sunrise"] < current["dt"] < current["sunset"] else None), "U{:2.0f}", "U -", Colour_spec("seq", "Purples", [0, 1, 5, 7, 9]) ) elif select == "now": res_o = Result( datetime.fromtimestamp(dat["current"]["dt"]).strftime("🕛 %d%H%MJ"), "{}", None, None ) elif select == "sunrise": res_o = Result( datetime.fromtimestamp(dat["current"]["sunrise"]).strftime("☀↑ %H:%M:%S"), "{}", None, None ) elif select == "sunset": res_o = Result( datetime.fromtimestamp(dat["current"]["sunset"]).strftime("☀↓ %H:%M:%S"), "{}", None, None ) elif select == "rain": # Actual rain and probability to be figured out later. cloud_level = dat["current"]["clouds"] if cloud_level > 0: res_o = Result( cloud_level, "☁ {:2d}%", None, Colour_spec("seq", "Blues", [0, 30, 70, 100]) ) else: res_o = Result(0, "☁⃠", None, None) elif select == "status": res_o = Result(dat["current"]["weather"][0]["main"], "{}", "", None) elif select == "hilo": time_horizon = 12 if 5 < datetime.fromtimestamp(dat["current"]["dt"]).hour < 17: maxloc, maxtemp = max(enumerate(dat["hourly"][:time_horizon]), key=lambda x: x[1]["temp"]) mintemp = min(dat["hourly"][maxloc : maxloc + time_horizon], key=lambda x: x["temp"]) res_o = Result( [(maxtemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(maxtemp["dt"]).hour, (mintemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(mintemp["dt"]).hour,], ('<span background="#801A00">↑ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span background="#005266">↓ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "", None ) else: minloc, mintemp = min(enumerate(dat["hourly"][:time_horizon]), key=lambda x: x[1]["temp"]) maxtemp = max(dat["hourly"][minloc : minloc + time_horizon], key=lambda x: x["temp"]) res_o = Result( [(mintemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(mintemp["dt"]).hour, (maxtemp["temp"] * units.kelvin).to("degC").magnitude, datetime.fromtimestamp(maxtemp["dt"]).hour], ('<span background="#005266">↓ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span background="#801A00">↑ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "", None ) return res_o if __name__ == "__main__": home = "~/Documents/Junk/weather-archives/" select = ((len(sys.argv) >= 2 and sys.argv[1]) or os.getenv("BLOCK_INSTANCE") or "temperature") res = collect_owm_data(load(home), select) res.print_results() sys.exit(res.return_code())
Raw
Repaste
python
default
isoraqathedh
2022.05.03 04:48:22
#!/usr/bin/python from datetime import datetime, timedelta import json from os import getenv import colorlover as colours import pathlib import sys # Data types: class Colour_spec: scale_type = "" scale_name = "" reverse = False bounds = [] def __init__(self, s_type, s_name, bounds, rev=False): self.scale_type = s_type self.scale_name = s_name self.bounds = bounds self.reverse = rev def is_urgent(self, value): return not self.bounds[0] <= value <= self.bounds[-1] def colour_of(self, value): # We want the gaps between, so we shorten the list a little. # The first colour is just a placeholder. selected_scale = [(255, 255, 255)] + colours.to_numeric( colours.scales [str(len(self.bounds) - 1)] [self.scale_type] [self.scale_name] [::(-1 if self.reverse else 1)]) if not self.is_urgent(value): for (bound, (red, green, blue)) in zip(self.bounds, selected_scale): if bound >= value: luminanceA = ((0.2126*red/255) + (0.7152*green/255) + (0.0722*blue/255)) colour_hex = "#{red:02X}{green:02X}{blue:02X}".format( red=round(red), green=round(green), blue=round(blue)) if luminanceA > 0.5: # Basic colour foreground = colour_hex background = None else: # Reversed for visibility foreground = "#FFFFFF" background = colour_hex break else: foreground = None background = None return (foreground, background) # Colour specs: colour_specs = { "temperature": Colour_spec("div", "RdBu", [7, 12, 17, 22, 27, 32], True), "wind": Colour_spec("seq", "Greens", [0, 12, 30, 40, 62, 87, 117], False), } # Result organisation and formatting class Result: val = None fmt = "" null_string = None colours = None force_urgent = False def __init__(self, val, fmt, null_string, colours=None, force_urgent=False): self.val = val self.fmt = fmt self.null_string = null_string self.colours = colours self.force_urgent = force_urgent def value_string(self): return (self.fmt.format(self.val) if self.val is not None else self.null_string) def return_code(self): return (33 if self.force_urgent or (self.colours and self.val and self.colours.is_urgent(self.val)) else 0) def colour(self): if self.val is None: return ("#999999", "#222222") if self.colours: return self.colours.colour_of(self.val) else: return (None, None) def print_results(self): print(self.value_string()) print(self.value_string()) fg, bg = self.colour() if fg: print(fg) if bg: print(bg) # Processing the JSON file: def load(home, prefix): for filename in sorted( pathlib.PosixPath(home).expanduser().glob(f"{prefix}_*.json"), key=lambda x: x.stat().st_ctime, reverse=True): recentp = True with open(filename) as f: try: res = json.load(f) if not recentp: res["recent"] = False return res except json.decoder.JSONDecodeError: recentp = False class Extractor(): def compute_result(self, dat, select): raise NotImplementedError("Please Implement this method") class OWM_extractor(Extractor): file_prefix = "owm" def compute_result(self, dat, select): if select == "temperature": return self._temperature(dat) elif select == "apparent_temperature": return self._apparent_temperature(dat) elif select == "wind_speed": return self._wind_speed(dat) elif select == "wind_gust": return self._wind_gust(dat) elif select == "wind_deg": return self._wind_deg(dat) elif select == "pressure": return self._pressure(dat) elif select == "visibility": return self._visibility(dat) elif select == "humidity": return self._humidity(dat) elif select == "UV": return self._UV(dat) elif select == "now": return self._now(dat) elif select == "sunrise": return self._sunrise(dat) elif select == "sunset": return self._sunset(dat) elif select == "rain": return self._rain(dat) elif select == "status": return self._status(dat) elif select == "hilo": return self._hilo(dat) else: return None # @supports_selector(self, "temperature") def _temperature(self, dat): return Result( dat["current"]["temp"] - 273.15, # K → °C "{:5.2f}°C", "--.--°C", colour_specs["temperature"] ) def _apparent_temperature(self, dat): apparent = dat["current"]["feels_like"] - 273.15 # K → °C real = dat["current"]["temp"] - 273.15 # K → °C return Result( apparent if abs(apparent - real) > 2 else None, "({:5.2f}°C)", "(--.--°C)", colour_specs["temperature"] ) def _wind_speed(self, dat): return Result( dat["current"]["wind_speed"] * 3.6, # m/s → km/h "{:5.1f} km/h", "---.- km/h", colour_specs["wind"] ) def _wind_gust(self, dat): return Result( (dat["current"]["wind_gust"] * 3.6 # m/s → km/h if "wind_gust" in dat["current"] and abs(dat["current"]["wind_gust"] - dat["current"]["wind_speed"]) >= 5 else None), "({:5.1f} km/h)", "(---.- km/h)", colour_specs["wind"] ) def _wind_deg(self, dat): return Result(dat["current"]["wind_deg"], "{:03d}°", "---°", None) def _pressure(self, dat): Result( dat["current"]["pressure"], "{:4d} hPa", "---- hPa", Colour_spec("div", "RdBu", [990, 995, 1000, 1005, 1010, 1015, 1020, 1025, 1050]) ) def _visibility(self, dat): return Result( (dat["current"]["visibility"] / 1000 # m → km if dat["current"]["visibility"] < 10000 else None), "{:4.2f} km", "-.-- km", Colour_spec("seq", "YlOrBr", [0.5, 3, 5, 10], True) ) def _humidity(self, dat): return Result( dat["current"]["humidity"], "{:3.0f}%", "---%", Colour_spec("div", "RdBu", [30, 50, 70, 80, 90, 95]) ) def _UV(self, dat): current = dat["current"] return Result( (current["uvi"] if current["uvi"] >= 0.5 and current["sunrise"] < current["dt"] < current["sunset"] else None), "U{:2.0f}", "U -", Colour_spec("seq", "Purples", [0, 1, 5, 7, 9], False) ) def _now(self, dat): time = datetime.fromtimestamp(dat["current"]["dt"]) now = datetime.now() ages = [now - timedelta(minutes=i) for i in [0, 10, 30, 60, 2 * 60, 24 * 60]] ages.reverse() return Result(time, "🕒 {:%d%H%M}J", None, Colour_spec("seq", "Oranges", ages, True)) def _sunrise(self, dat): sunrise = datetime.fromtimestamp(dat["current"]["sunrise"]) return Result(sunrise, "☀↑ {:%H:%M:%S}", None, None) def _sunset(self, dat): sunset = datetime.fromtimestamp(dat["current"]["sunset"]) return Result(sunset, "☀↓ {:%H:%M:%S}", None, None) def _rain(self, dat): rainingp = "rain" in dat["current"].keys() cloud_level = dat["current"]["clouds"] if rainingp: rain_level = dat["current"]["rain"]["1h"] colours = Colour_spec("seq", "Blues", [0, 1, 5, 10, 20, 30, 50, 70], False) res_o = Result(rain_level, "☔ {:5.2f} mm/h", None, colours) elif cloud_level > 0: colours = Colour_spec("seq", "Blues", [0, 30, 70, 100], False) res_o = Result(cloud_level, "☁ {:2d}%", None, colours) else: res_o = Result(0, "☁⃠", None, None) return res_o def _status(self, dat): icon_code = dat["current"]["weather"][0]["icon"] description = dat["current"]["weather"][0]["main"] try: icon = { "01d": "☀", "01n": "☽", "02d": "🌤", "02n": "🌤 (☽)", "03d": "🌥", "03n": "🌥 (☽)", "04d": "☁", "04n": "☁", "09d": "☔", "09n": "☔", "10d": "☀☔", "10n": "☽☔", "11d": "🌩", "11n": "🌩", "13d": "☃", "13n": "☃", "50d": "🌫", "50n": "🌫", }[icon_code] pass except IndexError: icon = "-" return Result(f"{icon} {description}", "{}", "", None) def _hilo(self, dat): time_horizon = 12 if 0 < datetime.fromtimestamp(dat["current"]["dt"]).hour < 12: maxloc, maxtemp = max(enumerate(dat["hourly"][5 : 5 + time_horizon]), key=lambda x: x[1]["temp"]) mintemp = min(dat["hourly"][maxloc + 1 : maxloc + time_horizon + 1], key=lambda x: x["temp"]) res_o = Result( [maxtemp["temp"] - 273.15, datetime.fromtimestamp(maxtemp["dt"]).hour, mintemp["temp"] - 273.15, datetime.fromtimestamp(mintemp["dt"]).hour,], ('<span foreground="#E3694D">↑ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span foreground="#00BAE7">↓ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "", None ) else: minloc, mintemp = min(enumerate(dat["hourly"][:time_horizon]), key=lambda x: x[1]["temp"]) maxtemp = max(dat["hourly"][minloc : minloc + time_horizon], key=lambda x: x["temp"]) res_o = Result( [mintemp["temp"] - 273.15, datetime.fromtimestamp(mintemp["dt"]).hour, maxtemp["temp"] - 273.15, datetime.fromtimestamp(maxtemp["dt"]).hour], ('<span foreground="#00BAE7">↓ {0[0]:4.1f}°C {0[1]:02d}h</span> ' '<span foreground="#E3694D">↑ {0[2]:4.1f}°C {0[3]:02d}h</span>'), "", None ) return res_o class AQI_extractor(Extractor): file_prefix = "aqi" def compute_result(self, dat, select): if select == "aqi": return self._aqi(dat) elif select == "pm25": return self._pm25(dat) else: return None def _aqi(self, dat): return Result(dat["data"]["aqi"], "∴ {:3d}", "∴ ---", Colour_spec("seq", "Oranges", [0, 26, 51, 101, 200])) def _pm25(self, dat): return Result(dat["data"]["iaqi"]["pm25"]["v"], "PM2½ {:2d} µg/m³", "PM2½ -- µg/m³", None) if __name__ == "__main__": home = "~/Documents/Junk/weather-archives/" select = ((len(sys.argv) >= 2 and sys.argv[1]) or getenv("BLOCK_INSTANCE") or "temperature") for extractor in [OWM_extractor(), AQI_extractor()]: res = extractor.compute_result( load(home, extractor.file_prefix), select) if res: break res.print_results() sys.exit(res.return_code())
Raw
Repaste