Skip to content

Cleanup and annotate #60

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 1, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions lib/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion lib/stock/stockquote.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
97 changes: 46 additions & 51 deletions lib/weather/weather.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,38 +4,39 @@
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:
icon.update({icn["OWMCode"]: WeatherIcon(icn['OWMCode'], icn['Description'], icn['fontcode'], icn['font'])})
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


Expand All @@ -46,14 +47,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")
Expand All @@ -80,7 +81,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')
Expand All @@ -91,7 +92,8 @@ 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):

def get_current_location(self) -> Dict[str, str]:
url = 'http://ipinfo.io/json'
response = self.run_non_async_request(url)
return response.json()
Expand All @@ -100,9 +102,9 @@ 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] = 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:
Expand All @@ -114,15 +116,8 @@ async def url_builder(self, location=None, zipcode=None, current_location=False)
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']
Expand Down Expand Up @@ -180,89 +175,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):
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:
Expand All @@ -272,14 +267,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
Expand All @@ -293,5 +288,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
4 changes: 1 addition & 3 deletions matrix/matrix.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,12 @@
from abc import abstractmethod
import asyncio
import functools
import configparser
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
)
Expand Down
Loading