91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
import urequests # type: ignore
|
|
import json
|
|
|
|
from app.classes import Weather, Location, CurrentCondition
|
|
from app.display.neopixel_64x64 import NeoPixel_64x64
|
|
import app.utils.colors as colors
|
|
import app.display.fonts as fonts
|
|
from app.utils import URLEncoder
|
|
|
|
|
|
API_KEY = '3545ce42d0ba436e8dc164532250410'
|
|
ACTUAL_WEATHER_URL = 'http://api.weatherapi.com/v1/current.json?key={API_KEY}&q={city}&aqi=yes&lang={lang}'
|
|
|
|
class Weather_Checker():
|
|
|
|
def __init__(self, display: NeoPixel_64x64):
|
|
self.display = display
|
|
self.display.set_font( font=fonts.font_5x7)
|
|
|
|
def mock_weather_data(self):
|
|
filepath = 'restapi/mock-weather.json'
|
|
try:
|
|
with open(filepath, 'r', encoding='utf-8') as file:
|
|
return json.load(file)
|
|
except OSError: # Use OSError instead of FileNotFoundError
|
|
print(f"Error: File '{filepath}' not found.")
|
|
return None
|
|
except FileNotFoundError:
|
|
print(f"Error: File '{filepath}' not found.")
|
|
return None
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error: Invalid JSON in '{filepath}': {e}")
|
|
return None
|
|
|
|
|
|
def actual_weather(self, city, lang, test_mode=False) -> Weather:
|
|
weather_url = ACTUAL_WEATHER_URL.format(API_KEY=API_KEY, city=city, lang=lang)
|
|
|
|
if not test_mode:
|
|
|
|
print(f'query: {weather_url}')
|
|
r = urequests.get(weather_url)
|
|
|
|
print('Status-Code:', r.status_code)
|
|
json_resp = r.json()
|
|
|
|
r.close()
|
|
else:
|
|
print('Status-Code: test_mode')
|
|
# json_resp = WEATHER_QUERY_MOCK
|
|
json_resp = self.mock_weather_data()
|
|
|
|
loc = Location(**json_resp['location'])
|
|
cc = CurrentCondition(**json_resp['current'])
|
|
|
|
weather = Weather(location=loc, current=cc)
|
|
|
|
return weather
|
|
|
|
|
|
def check(self, city:str, lang:str, test_mode: bool = False):
|
|
try:
|
|
self.display.clear()
|
|
|
|
city = URLEncoder.encode_utf8(city) # url_encode
|
|
w_resp: Weather = self.actual_weather(city=city, lang=lang, test_mode=test_mode)
|
|
|
|
ypos = 0
|
|
self.display.write_text(f'{str(w_resp.location.name)}', 0, ypos, color=colors.RAINBOW[0])
|
|
ypos += self.display.font_height + 1
|
|
self.display.write_text(f'{str(w_resp.location.region)}', 0, ypos, color=colors.RAINBOW[0])
|
|
ypos += self.display.font_height + 1
|
|
self.display.write_text(f'{str(w_resp.location.localtime)[:10]}', 0, ypos, color=colors.RAINBOW[3])
|
|
ypos += self.display.font_height + 1
|
|
|
|
self.display.write_text(f'{str(w_resp.current.temp_c)}°C', 0, ypos, color=colors.RAINBOW[1])
|
|
ypos += self.display.font_height + 1
|
|
self.display.write_text(f'{str(w_resp.current.condition.text)}', 0, ypos, color=colors.RAINBOW[2])
|
|
ypos += self.display.font_height + 1
|
|
self.display.write_text(f'upd:{str(w_resp.current.last_updated)[-5:]}', 0, ypos, color=colors.RAINBOW[3])
|
|
ypos += self.display.font_height + 1
|
|
self.display.write_text(f'cur:{str(w_resp.location.localtime)[-5:]}', 0, ypos, color=colors.RAINBOW[4])
|
|
|
|
# unten rechts in die Ecke
|
|
ypos = self.display.MATRIX_HEIGHT - self.display.font_height - 1
|
|
|
|
self.display.write_text('done.', 38, ypos, color=colors.NEON_GREEN)
|
|
except OSError as e:
|
|
print(f'Error: connection closed - {e}')
|
|
finally:
|
|
print('finally done.') |