diff --git a/lib/binary_build.sh b/lib/binary_build.sh index c3797ca..9f4f0d1 100755 --- a/lib/binary_build.sh +++ b/lib/binary_build.sh @@ -1,5 +1,5 @@ #!/bin/bash -VERSION="0.4.0" +VERSION="0.5.0" # Can't put full folders echo "Building the binary" diff --git a/lib/run.py b/lib/run.py index 3313ee2..fa5a450 100755 --- a/lib/run.py +++ b/lib/run.py @@ -5,6 +5,7 @@ import requests import logging import aiohttp +from requests.models import Response stream_formatter = logging.Formatter( "%(levelname)s:%(asctime)s:%(module)s:%(message)s" @@ -61,22 +62,26 @@ def __init__(self, config): else: self.logger.setLevel(10) self.logger.debug(f"Runner logger is set to {self.logger.getEffectiveLevel()}") - def run_non_async_request(self, url): + + def run_non_async_request(self, url) -> Response: response = requests.get(url) return response - async def get_data(self, url, headers: Dict[str, str]={}): + + async def get_data(self, url, headers: Dict[str, str]={}) -> Dict: self.logger.debug(f'Getting data with URL {url}') async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: data = await resp.json() return data - async def get_non_json_data(self, url, headers: Dict[str, str]={}): + + async def get_non_json_data(self, url, headers: Dict[str, str]={}) -> Dict: self.logger.debug(f'Getting data with URL {url}') async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: data = await resp.text() return data - async def get_img(self, url, headers: Dict[str, str]={}): + + async def get_img(self, url, headers: Dict[str, str]={}) -> Dict: self.logger.debug(f'Getting data with URL {url}') async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: diff --git a/lib/stock/stockquote.py b/lib/stock/stockquote.py index 1002928..835aa82 100755 --- a/lib/stock/stockquote.py +++ b/lib/stock/stockquote.py @@ -26,7 +26,7 @@ async def run(self) -> Dict: """ returns current stock Quotes and Price """ - self.logger.info("Running Stock Data") + self.logger.info("Running Stock Api") symbol = self.config.get('symbol') api_data = await self.get_data(self.url_builder(symbol=symbol)) api_data['symbol'] = symbol diff --git a/lib/weather/weather.py b/lib/weather/weather.py index e27a057..dec13df 100755 --- a/lib/weather/weather.py +++ b/lib/weather/weather.py @@ -1,17 +1,22 @@ #!/usr/bin/env python3 +import asyncio +from logging import currentframe from lib.run import Runner, Caller +from lib.asynclib import make_async import sys import os import json -from typing import Dict, Tuple +from typing import Dict, Tuple, List from datetime import datetime +from datetime import timedelta import csv -def get_weather_csv(): +def get_weather_csv() -> List[Dict[str, str]]: csv_path = '/etc/ohmyoled/ecIcons_utf8.csv' return list(csv.DictReader(open(csv_path))) -def build_weather_icons(): + +def build_weather_icons() -> str: csv = get_weather_csv() icon = {} for icn in csv: @@ -19,23 +24,23 @@ def build_weather_icons(): return icon class WeatherIcon(): - def __init__(self, owm_id, description, fontcode, font) -> None: + def __init__(self, owm_id: str, description: str, fontcode: str, font: str) -> None: self.owm_id = owm_id self.description = description self.fontcode = fontcode self.font = font @property - def get_owm_id(self): + def get_owm_id(self) -> str: return self.owm_id @property - def get_description(self): + def get_description(self) -> str: return self.description @property - def get_fontcode(self): + def get_fontcode(self) -> str: return self.fontcode @property - def get_font(self): + def get_font(self) -> str: return self.font @@ -46,14 +51,14 @@ class WeatherApi(Runner): To parse the config file and Run Data """ - def __init__(self, config): + def __init__(self, config) -> None: super().__init__(config) self.weather = self.config['weather'] try: if "open_weather_token" in self.config['basic']: - self.token = self.config['basic'].get('open_weather_token') + self.token: str = self.config['basic'].get('open_weather_token') else: - self.token = os.environ['WEATHERTOKEN'] + self.token: str = os.environ['WEATHERTOKEN'] except KeyError: self.logger.critical("No Weather Token") sys.exit("No Weather Token") @@ -80,7 +85,7 @@ async def get_long_and_lat(self, location: str=None, zipcode: int=None) -> Tuple self.logger.debug("Getting Lat and Long") try: if location: - self.logger.debug("Getting Longitude and Latitude") + self.logger.debug("Computing Longitude and Latitude") url = f'http://api.openweathermap.org/data/2.5/weather?q={location}&appid={self.token}' response = await self.get_data(url) lon = response.get('coord').get('lon') @@ -91,7 +96,9 @@ async def get_long_and_lat(self, location: str=None, zipcode: int=None) -> Tuple except Exception as e: self.logger.critical(e) sys.exit("No City Found") - def get_current_location(self): + + @make_async + def get_current_location(self) -> Dict[str, str]: url = 'http://ipinfo.io/json' response = self.run_non_async_request(url) return response.json() @@ -100,32 +107,26 @@ async def url_builder(self, location=None, zipcode=None, current_location=False) """ Builds Url to poll the Api """ - self.logger.debug("Building url...") + self.logger.debug("Building Weather url...") if current_location: - ip_json = self.get_current_location() + ip_json: Dict[str, str] = await self.get_current_location() lon, lat = ip_json['loc'].split(',')[1], ip_json['loc'].split(',')[0] url = f"https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&appid={self.token}&units={self.weather.get('format')}" elif location: lon, lat = await self.get_long_and_lat(location) url = f"https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&appid={self.token}&units={self.weather.get('format')}" else: - ip_json = self.get_current_location() + ip_json = await self.get_current_location() lon, lat = ip_json['loc'].split(',')[1], ip_json['loc'].split(',')[0] url = f"https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&appid={self.token}&units={self.weather.get('format')}" return url - async def run(self): - """ - Get Args - parse args - Build URL - make request - return Json - """ - self.logger.info("Using to get Weather") + async def run(self) -> Dict: + self.logger.info("Running Api for Weather") args = await self.parse_args() api_data = await self.get_data(args) - api_data['name'] = self.get_current_location()['city'] + current_data = await self.get_current_location() + api_data['name'] = current_data['city'] return api_data class Weather(Caller): @@ -180,89 +181,89 @@ def __repr__(self) -> str: return f"Weather(\n{joined_attrs})" @property - def get_wind_speed(self): + def get_wind_speed(self) -> int: return self._wind_speed - def set_wind_speed(self, speed): + def set_wind_speed(self, speed: int) -> None: self._wind_speed = speed @property - def get_daily(self): + def get_daily(self) -> Dict[str, str]: return self._daily @property - def get_wind_deg(self): + def get_wind_deg(self) -> int: return self._wind_deg @property - def get_precipitation(self): - return self._pop + def get_precipitation(self) -> int: + return self._pop * 100 @property - def get_uv(self): + def get_uv(self) -> int: return self._uv - def set_place(self, place): + def set_place(self, place: str) -> None: self._place = place @property - def get_place(self): + def get_place(self) -> str: return self._place - def set_weather(self, weather): + def set_weather(self, weather: Dict[str, str]) -> None: self._weather = weather @property - def get_weather(self): + def get_weather(self) -> Dict[str, str]: return self._weather - def set_conditions(self, conditions): + def set_conditions(self, conditions: str) -> None: self._conditions = conditions @property - def get_conditions(self): + def get_conditions(self) -> str: return self._conditions - def set_weather_icon(self, icon): + def set_weather_icon(self, icon: str) -> None: self._weather_icon = icon @property - def get_weather_icon(self): + def get_weather_icon(self) -> str: return self._weather_icon - def set_temp(self, temp): + def set_temp(self, temp: int) -> None: self._temp = temp @property - def get_temp(self): + def get_temp(self) -> int: return self._temp - def set_feels_like(self, feels): + def set_feels_like(self, feels: int): self._feels_like = feels @property - def get_feels_like(self): + def get_feels_like(self) -> int: return self._feels_like - def set_min_temp(self, temp): + def set_min_temp(self, temp: int) -> None: self._min_temp = temp @property - def get_min_temp(self): + def get_min_temp(self) -> int: return self._min_temp - def set_max_temp(self, temp): + def set_max_temp(self, temp: int) -> None: self._max_temp = temp @property - def get_max_temp(self): + def get_max_temp(self) -> int: return self._max_temp - def set_humidity(self, humidity): + def set_humidity(self, humidity: int) -> None: self._humidity = humidity @property - def get_humidity(self): + def get_humidity(self) -> None: return self._humidity def set_wind(self, wind: Dict) -> None: @@ -272,14 +273,14 @@ def set_wind(self, wind: Dict) -> None: def get_wind(self) -> Dict: return self._wind - def set_time(self, time) -> None: + def set_time(self, time: int) -> None: self._time = datetime.fromtimestamp(time) @property def get_time(self) -> datetime: return self._time - def set_sunrise(self, time) -> None: + def set_sunrise(self, time: int) -> None: self._sunrise = datetime.fromtimestamp(time) @property @@ -293,5 +294,5 @@ def set_sunset(self, time) -> None: def get_sunset(self) -> datetime: return self._sunset - def calculate_duration_of_daylight(self): + def calculate_duration_of_daylight(self) -> timedelta: return self._sunset - self._time diff --git a/main.py b/main.py index e18cb54..405f7bc 100755 --- a/main.py +++ b/main.py @@ -2,22 +2,18 @@ import asyncio import configparser -import os +import time from rgbmatrix import ( RGBMatrixOptions, RGBMatrix ) from lib.weather.weather import WeatherApi from matrix.stock.stockmatrix import StockMatrix -from matrix.stock.historicalstockmatrix import HistoricalStockMatrix -from lib.stock.stocks import StockApi, Stock -from lib.sports.sports import SportApi, Sport -from matrix.matrix import Matrix +from lib.stock.stocks import StockApi +from lib.sports.sports import SportApi from matrix.time import TimeMatrix from matrix.weathermatrix import WeatherMatrix from matrix.sport.sportmatrix import SportMatrix -from abc import abstractmethod -import requests import logging @@ -32,6 +28,8 @@ logger.addHandler(sh) logger.addHandler(filehandler) +class OledExecption(Exception): + pass class Main(): def __init__(self, config) -> None: self.config = config @@ -40,6 +38,7 @@ def __init__(self, config) -> None: self.logger.setLevel(self.config['basic'].getint('loglevel')) else: self.logger.setLevel(10) + self.poll = None self.logger.debug(f"Logger is set to {self.logger.getEffectiveLevel()}") def parse_config_file(self): module = {} @@ -100,26 +99,50 @@ async def init_matrix(self, matrix): verified_modules.append(SportMatrix(matrix, modules['sport'], logger)) self.logger.info("Initalized matrixes") return verified_modules + + async def run_matrix_worker(self, matrix, polled_data): + self.logger.debug("Starting Worker") + # Need to make sure these need to be coroutines + matrix.render(polled_data) + + async def poll_api_worker(self, matrix): + polled_data = await matrix.poll_api() + return polled_data - async def main_run(self): - self.logger.info("Starting OhMyOled") - matrix = RGBMatrix(options=self.poll_rgbmatrix()) - self.logger.debug("Built Options for RGBMatrix") - matrixes = await self.init_matrix(matrix) - self.logger.info("Starting Matrixes...") - while True: - for matrix in matrixes: - poll = await matrix.poll_api() - matrix.render(poll) + async def main_run(self, loop): + try: + self.logger.info("Starting OhMyOled") + matrix = RGBMatrix(options=self.poll_rgbmatrix()) + self.logger.debug("Built Options for RGBMatrix") + matrixes = await self.init_matrix(matrix) + self.logger.info("Starting Matrixes...") + first_poll = True + while True: + for index, matrix in enumerate(matrixes): + matrix_start_time = time.perf_counter() + if first_poll: + self.poll = await matrix.poll_api() + first_poll = False + if index + 1 >= len(matrixes): + tasks = [asyncio.create_task(matrix.render(self.poll, loop)), asyncio.create_task(matrixes[0].poll_api())] + else: + tasks = [asyncio.create_task(matrix.render(self.poll, loop)), asyncio.create_task(matrixes[index+1].poll_api())] + _, self.poll = await asyncio.gather(*tasks) + matrix_finish_time = time.perf_counter() + logger.info(f"{matrix} rendered for {matrix_finish_time - matrix_start_time:0.4f}s") + except Exception as E: + logger.error(E) + loop.stop() if __name__ == "__main__": config = configparser.ConfigParser() logger.info("Pulling configuration /etc/ohmyoled/ohmyoled.conf") config.read('/etc/ohmyoled/ohmyoled.conf') main = Main(config) + loop = asyncio.get_event_loop() try: - loop.create_task(main.main_run()) + loop.create_task(main.main_run(loop)) loop.run_forever() except KeyboardInterrupt: logger.critical("Key Interrupt") diff --git a/matrix/__init__.py b/matrix/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/matrix/matrix.py b/matrix/matrix.py index 923e5bf..a4f19d3 100755 --- a/matrix/matrix.py +++ b/matrix/matrix.py @@ -1,18 +1,15 @@ #!/usr/bin/python3 # ABS_Matrix -> Matrix_Module -# -# -# from abc import abstractmethod -import configparser +import asyncio +import functools import logging from sys import exec_prefix -from PIL import Image, ImageDraw, ImageFont +from PIL import Image, ImageDraw from typing import Deque, Tuple, List from collections import deque from rgbmatrix import ( - RGBMatrixOptions, RGBMatrix, graphics ) @@ -134,8 +131,18 @@ def image_resize(self, width, height) -> None: self.set_image(self.image.resize((width, height), Image.ANTIALIAS)) self.set_draw(ImageDraw.Draw(self.image)) - def render_image(self, xoffset=0, yoffset=0): - self.matrix.SetImage(self.get_image, offset_y=yoffset, offset_x=xoffset) + async def render_image(self, loop=None, xoffset=0, yoffset=0): + if not loop: + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + functools.partial( + self.matrix.SetImage, + self.get_image, + offset_x=xoffset, + offset_y=yoffset + ) + ) def draw_rectangle(self, position: List[Tuple]): """ diff --git a/matrix/sport/__init__.py b/matrix/sport/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/matrix/sport/sportmatrix.py b/matrix/sport/sportmatrix.py index 70d271d..49b01cb 100755 --- a/matrix/sport/sportmatrix.py +++ b/matrix/sport/sportmatrix.py @@ -1,27 +1,24 @@ #!/usr/bin/env python3 -import asyncio +from logging import Logger import os import time import urllib.request from datetime import datetime from typing import List, Dict, Tuple from collections import deque -from datetime import datetime, timedelta from PIL import ImageFont, Image, ImageDraw from lib.sports.sports import Sport from matrix.matrix import Matrix -from lib.sports.baseball.baseball import Baseball -from lib.sports.basketball.basketball import Basketball -from lib.sports.hockey.hockey import Hockey -from matrix.sport.team_mapping import BASEBALL_TEAMS, BASKETBALL_TEAMS class SportMatrix(Matrix): - def __init__(self, matrix, api, logger) -> None: + def __init__(self, matrix, api: Sport, logger: Logger) -> None: self.matrix = matrix self.api = api self.logger = logger - async def poll_api(self): + def __str__(self) -> str: + return "SportMatrix" + async def poll_api(self) -> Sport: return Sport(await self.api.run()) def baseball_divisions(self, standings: List[Dict]) -> List[str]: @@ -35,8 +32,7 @@ def baseball_divisions(self, standings: List[Dict]) -> List[str]: return list(american_queue), list(national_queue) def determine_nextgame(self, nextgame_api): - now = datetime.now() - status = ("FT", "ABD") + status: Tuple = ("FT", "ABD") for game in nextgame_api: if "IN" in game['status']: self.logger.debug(f"In Game") @@ -58,13 +54,13 @@ def home_team(self, nextgame_data): home.update(nextgame_data['score']['home']) return home - def get_logo(self, logo_url, name): - file_name = f"/tmp/{name}.png" + def get_logo(self, logo_url: str, name: str) -> str: + file_name: str = f"/tmp/{name}.png" if not os.path.isfile(file_name): urllib.request.urlretrieve(logo_url, file_name) return file_name - def build_in_game_image(self, nextgame): + def build_in_game_image(self, nextgame: Dict): middle_image = self.make_new_image((34,16)) middle_draw = ImageDraw.Draw(middle_image) font = ImageFont.truetype("/usr/share/fonts/fonts/04B_03B_.TTF", 8) @@ -72,25 +68,27 @@ def build_in_game_image(self, nextgame): self.logger.debug(f"status: {status}") score = (nextgame['teams']['home']['total'], nextgame['teams']['away']['total']) self.logger.debug(f"Score: {score}") - middle_draw.multiline_text((12,0), f" {status}\n{score[0]}-{score[1]}", font=font) + middle_draw.multiline_text((12,0), f"{status}\n{score[0]}-{score[1]}", font=font) return middle_image, (15, 0) + def build_finished_game_image(self, nextgame): middle_image = self.make_new_image((34,16)) middle_draw = ImageDraw.Draw(middle_image) font = ImageFont.truetype("/usr/share/fonts/fonts/04B_03B_.TTF", 8) status = nextgame['status'] score = (nextgame['teams']['home']['total'], nextgame['teams']['away']['total']) - middle_draw.multiline_text((12,0), f" {status}\n{score[0]}-{score[1]}", font=font) + middle_draw.multiline_text((12,0), f"{status}\n{score[0]}-{score[1]}", font=font) return middle_image, (15, 0) - def check_offseason(self, api): + def check_offseason(self, api) -> bool: try: start_time, end_time = api.get_timestamps[0][1], api.get_timestamps[-1][1] if datetime.fromtimestamp(start_time) <= datetime.now() <= datetime.fromtimestamp(end_time): return True except Exception: return False - def build_next_game_image(self, nextgame): + + def build_next_game_image(self, nextgame: Dict): middle_image = self.make_new_image((34,16)) middle_draw = ImageDraw.Draw(middle_image) font = ImageFont.truetype("/usr/share/fonts/fonts/04B_03B_.TTF", 8) @@ -115,7 +113,7 @@ def build_home_away_image(self, nextgame): away_logo.thumbnail((16,16)) return (home_logo, (-2,0)), (away_logo, (50, 0)) - def build_middle_nextgame(self, api): + def build_middle_nextgame(self, api) -> Image: nextgame = self.determine_nextgame(api.next_game) if "IN" in nextgame['status']: return self.build_in_game_image(nextgame) @@ -124,7 +122,7 @@ def build_middle_nextgame(self, api): elif "FT" == nextgame['status']: return self.build_finished_game_image(nextgame) - def build_middle_image(self, api): + def build_middle_image(self, api) -> Image: nextgame = self.determine_nextgame(api.next_game) home_image, away_image = self.build_home_away_image(nextgame) middle_image = self.build_middle_nextgame(api) @@ -134,7 +132,7 @@ def build_middle_image(self, api): master_middle_image.paste(middle_image[0], middle_image[1]) return master_middle_image, (0,9) - def build_top_home_away_images(self, nextgame, xpos): + def build_top_home_away_images(self, nextgame: Dict, xpos: int) -> Tuple: font = ImageFont.truetype("/usr/share/fonts/fonts/04B_03B_.TTF", 8) top_home_image = self.make_new_image((22,8)) top_home_draw = ImageDraw.Draw(top_home_image) @@ -146,7 +144,7 @@ def build_top_home_away_images(self, nextgame, xpos): top_away_draw.text((-xpos,0), awayteam, font=font) return top_home_image, top_away_image - def build_top_image(self, api, xpos): + def build_top_image(self, api: Dict, xpos: int) -> Tuple: nextgame = self.determine_nextgame(api.next_game) master_top_image = self.make_new_image((64, 8)) home_image, away_image = self.build_top_home_away_images(nextgame, xpos) @@ -178,7 +176,7 @@ def build_standings_image(self, api, xpos) -> Tuple[int, int]: ) return standings_image, (0, 25) - def render(self, api): + async def render(self, api, loop): self.clear() self.reload_image() if 'baseball'in api.sport: @@ -199,16 +197,16 @@ def render(self, api): ) for image, position in images: self.paste_image(image, position) - self.render_image() + await self.render_image() xpos +=1 xpos_for_top += 1 if xpos_for_top == 100: xpos_for_top = 0 - time.sleep(3) if xpos == 1 else time.sleep(.03) + time.sleep(3) if xpos == 1 else time.sleep(.001) else: font = ImageFont.truetype("/usr/share/fonts/fonts/04b24.otf", 14) self.draw_multiline_text((0, 0), "Basketball\nOffseason", font=font) - self.render_image() + await self.render_image() time.sleep(30) if 'basketball' in api.sport: # Check Data if Offseason if yes Diplay Offseason, Otherwise Display Data @@ -233,7 +231,7 @@ def render(self, api): xpos_for_top += 1 if xpos_for_top == 100: xpos_for_top = 0 - time.sleep(3) if xpos == 1 else time.sleep(.05) + time.sleep(3) if xpos == 1 else time.sleep(.01) else: font = ImageFont.truetype("/usr/share/fonts/fonts/04b24.otf", 14) self.draw_multiline_text((0, 0), "Basketball\nOffseason", font=font) diff --git a/matrix/stock/__init__.py b/matrix/stock/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/matrix/stock/stockmatrix.py b/matrix/stock/stockmatrix.py index 7650acd..fba082a 100755 --- a/matrix/stock/stockmatrix.py +++ b/matrix/stock/stockmatrix.py @@ -12,7 +12,9 @@ def __init__(self, matrix, api, logger) -> None: self.matrix = matrix self.api = api self.logger = logger - + def __str__(self) -> str: + return "StockMatrix" + async def poll_api(self) -> Stock: return Stock(await self.api.run()) @@ -148,7 +150,7 @@ def render_lowest_price(self, api) -> None: fill=(255,0,0) ) - def render(self, api) -> None: + async def render(self, api, loop) -> None: self.logger.info("Started Render for Stock Matrix") self.clear() self.reload_image() @@ -157,5 +159,5 @@ def render(self, api) -> None: self.render_previous_close(api) self.render_highest_price(api) self.render_lowest_price(api) - self.render_image() + await self.render_image() time.sleep(30) \ No newline at end of file diff --git a/matrix/test.py b/matrix/test.py index 688e808..9f08ffd 100755 --- a/matrix/test.py +++ b/matrix/test.py @@ -1,4 +1,5 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +""" import time import sys import datetime @@ -70,6 +71,7 @@ scrolling_matrix.SetImage(master_image) # Image within an image """ +""" scrolling_font = ImageFont.truetype("/usr/share/fonts/fonts/04B_03B_.TTF", 8) scrolling_matrix = RGBMatrix(options=bottom_options) img1_text = Image.new("RGB", (22, 5)) @@ -176,4 +178,17 @@ time.sleep(.05) """ -time.sleep(30) \ No newline at end of file +#time.sleep(30) + +import asyncio +from setuptools import find_packages, setup +breakpoint() +#@make_async +def test1(arg1, arg2): + print(arg1, arg2) + +def test2(arg1, arg2): + print(arg1, arg2) + + +asyncio.run(test1("hello", "world")) \ No newline at end of file diff --git a/matrix/time.py b/matrix/time.py index f80c81b..7985206 100755 --- a/matrix/time.py +++ b/matrix/time.py @@ -16,21 +16,24 @@ def __init__(self, matrix, config) -> None: self.matrix = matrix self.config = config self.logger.debug("Time Matrix Initalized") - - def return_time(self, fmt: str): + + def __str__(self) -> str: + return "TimeMatrix" + + def return_time(self, fmt: str) -> datetime: return datetime.now().strftime(fmt) - async def poll_api(self): + async def poll_api(self) -> None: """ Function that does not poll since this a time """ - self.logger.debug("No Api call reqiured for time module") + self.logger.info("No Api call required for time module") return None - def build_fmt(self): + def build_fmt(self) -> str: return "%I:%M:%S %p" if TimeFormat.TWELEVE else "%H:%M:%S" - def render(self, poll): + async def render(self, poll: None, loop): # Build something that Loads in corner for all the modules loaded self.logger.info("Running Module TimeMatrix") counter = 0 @@ -41,6 +44,6 @@ def render(self, poll): self.set_draw(ImageDraw.Draw(self.get_image)) self.draw_text((3, 5), f"{self.return_time('%m/%d/%Y')}", font=font, fill=eval(self.config.get('color'))) self.draw_text((8, 16), f"{self.return_time('%I:%M:%S')}", font=font, fill=eval(self.config.get('color'))) - self.render_image() + await self.render_image() counter = counter + 1 time.sleep(1) \ No newline at end of file diff --git a/matrix/weathermatrix.py b/matrix/weathermatrix.py index c3bb83d..466a6cf 100755 --- a/matrix/weathermatrix.py +++ b/matrix/weathermatrix.py @@ -1,14 +1,10 @@ #!/usr/bin/env python3 -import asyncio import time -from typing import Dict +from typing import Dict, Tuple from datetime import datetime -import os -from PIL import Image -from PIL import ImageDraw from PIL import ImageFont -from matrix.matrix import Matrix, MatrixBase, FontException +from matrix.matrix import Matrix from lib.weather.weather import ( Weather, build_weather_icons @@ -21,9 +17,12 @@ def __init__(self, matrix, api: Dict, logger) -> None: self.logger = logger self.icons = build_weather_icons() + def __str__(self) -> str: + return "WeatherMatrix" async def poll_api(self) -> Weather: return Weather(await self.api.run()) - def get_temp_color(self, temp): + + def get_temp_color(self, temp: int) -> Tuple[int, int, int]: if temp >= 100: return (255, 12, 3) elif temp in range(70, 99): @@ -34,63 +33,69 @@ def get_temp_color(self, temp): return (0, 255, 255) else: return (0, 76, 255) - def render_temp(self, api): - font = ImageFont.truetype("/usr/share/fonts/retro_computer.ttf", 7) - metric = "\uf045" - self.draw_text((0, 10), "T:", font=font) - self.draw_text((10, 10), f"{str(int(api.get_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) - self.draw_text((30, 10), "R:", font=font) - self.draw_text((40, 10), f"{str(int(api.get_feels_like))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) - self.draw_text((1,20), f"H:", font=font) - self.draw_text((10, 20), f"{str(int(api.get_max_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) - self.draw_text((30, 20), f"L:", font=font) - self.draw_text((40, 20), f"{str(int(api.get_min_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) - def render_icon(self, api): - font = ImageFont.truetype("/usr/share/fonts/weathericons.ttf", 9) - owm_wxcode = int(api.get_weather[0]['id']) + def render_temp(self, api) -> None: + font: ImageFont = ImageFont.truetype("/usr/share/fonts/retro_computer.ttf", 7) + self.draw_text((0, 8), "T:", font=font) + self.draw_text((10, 8), f"{str(int(api.get_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) + self.draw_text((30, 8), "R:", font=font) + self.draw_text((40, 8), f"{str(int(api.get_feels_like))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) + self.draw_text((1, 18), f"H:", font=font) + self.draw_text((10, 18), f"{str(int(api.get_max_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) + self.draw_text((30, 18), f"L:", font=font) + self.draw_text((40, 18), f"{str(int(api.get_min_temp))}F", font=font, fill=self.get_temp_color(int(api.get_temp))) + + def render_icon(self, api: Weather) -> None: + font: ImageFont = ImageFont.truetype("/usr/share/fonts/weathericons.ttf", 9) + owm_wxcode: int = int(api.get_weather[0]['id']) if owm_wxcode in range(200,299): # Thunderstorm Class owm_icon = 200 - color = (254, 204, 1) + color: Tuple[int] = (254, 204, 1) elif owm_wxcode in range(300,399): # Drizzle Class owm_icon = 300 - color = (220,220,220) + color: Tuple[int] = (220,220,220) elif owm_wxcode in range(500,599): # Rain Class owm_icon = 500 - color = (108, 204, 228) + color: Tuple[int] = (108, 204, 228) elif owm_wxcode in range(600,699): # Snow Class owm_icon = 600 - color = (255,255,255) + color: Tuple[int] = (255,255,255) + elif owm_wxcode in range(700, 780): + owm_icon = 711 + color: Tuple[int] = (192, 192, 192) elif owm_wxcode == 800: # Sunny if api.get_sunset > datetime.now(): owm_icon = 800 - color = (220, 149, 3) + color: Tuple[int] = (220, 149, 3) else: owm_icon = 806 - color = (255,255,255) + color: Tuple[int] = (255,255,255) elif owm_wxcode in range(801,805): # Rain Class owm_icon = 801 - color = (220,220,220) + color: Tuple[int] = (220,220,220) else: owm_icon = owm_wxcode weather_icon = self.icons[str(owm_icon)] self.draw_text((50, 0), weather_icon.get_font, font, fill=color) - def render_location(self, api: Weather): + + def render_location(self, api: Weather) -> None: font = ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF",8) self.draw_text((2, 1), api.get_place, font, (0, 254, 0)) - def render_humidity (self, api: Weather): + + def render_humidity (self, api: Weather) -> None: font = ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF", 8) self.draw_text((2, 8), "H:", font) self.draw_text((10, 8), f"{api.get_humidity}%", font, fill=(7, 250, 246)) self.draw_text((27, 8), f"P:", font) self.draw_text((34, 8), f"{int(api.get_precipitation)}%", font, fill=(7, 250, 246)) - def render_wind(self, api: Weather): + + def render_wind(self, api: Weather) -> None: font = ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF", 8) speed = api.get_wind_speed deg = api.get_wind_deg @@ -98,26 +103,40 @@ def render_wind(self, api: Weather): self.draw_text((15, 15), f"{str(int(deg))}", font, fill=(201, 1, 253)) self.draw_text((30, 13), "\uf042", font=ImageFont.truetype("/usr/share/fonts/weathericons.ttf", 9), fill=(201, 1, 253)) self.draw_text((36, 15), f"{str(int(speed))}mph", font, fill=(201, 1, 253)) - def render_time(self, api: Weather): - font = ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF", 8) - sunrise = api.get_sunrise.strftime("%H:%M") - sunset = api.get_sunset.strftime("%H:%M") + + def render_time(self, api: Weather) -> None: + font: ImageFont = ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF", 8) + sunrise: datetime = api.get_sunrise.strftime("%H:%M") + sunset: datetime = api.get_sunset.strftime("%H:%M") self.draw_text((1, 18), "\uf058", font=ImageFont.truetype("/usr/share/fonts/weathericons.ttf", 11), fill=(255, 255, 0)) self.draw_text((7, 23), sunrise, font=font) self.draw_text((35, 18), "\uf044", font=ImageFont.truetype("/usr/share/fonts/weathericons.ttf", 11), fill=(255, 145, 0)) self.draw_text((40, 23), sunset, font=font) - def render(self, api: Weather): + def render_conditions(self, api: Weather, xpos: int) -> None: + self.draw_text((-xpos, 26), f"Conditions: {api.get_conditions}", font=ImageFont.truetype("/usr/share/fonts/04B_03B_.TTF", 8), fill=(255,255,255)) + async def render(self, api: Weather, loop) -> None: self.logger.info("Rendering Weather Matrix") self.logger.debug("Clearing Image") self.clear() self.logger.debug("Reloading Image in matrix") + xpos = 0 + self.logger.info("Loading Screen 1 of Matrix") + while xpos < 100: + self.reload_image() + self.render_temp(api) + self.render_icon(api) + self.render_location(api) + self.render_conditions(api, xpos) + xpos += 1 + await self.render_image() + time.sleep(3) if xpos == 1 else time.sleep(.05) self.reload_image() self.render_temp(api) self.render_icon(api) self.render_location(api) - self.logger.info("Loading Screen 1 of Matrix") - self.render_image() - time.sleep(30) + self.render_conditions(api, 0) + await self.render_image() + time.sleep(25) self.clear() self.logger.debug("Reloading Image in matrix") self.reload_image() @@ -127,6 +146,6 @@ def render(self, api: Weather): self.render_wind(api) self.render_time(api) self.logger.info("Loading Screen 2 of Matrix") - self.render_image() + await self.render_image() time.sleep(30)