mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-12 05:42:59 +00:00
Add more weather features
Added 6 hour and 3 day forecast. Removed refresh rate limit
This commit is contained in:
@@ -20,7 +20,7 @@
|
|||||||
"disable_hardware_pulsing": true,
|
"disable_hardware_pulsing": true,
|
||||||
"inverse_colors": false,
|
"inverse_colors": false,
|
||||||
"show_refresh_rate": true,
|
"show_refresh_rate": true,
|
||||||
"limit_refresh_rate_hz": 100
|
"limit_refresh_rate_hz": 0
|
||||||
},
|
},
|
||||||
"runtime": {
|
"runtime": {
|
||||||
"gpio_slowdown": 2
|
"gpio_slowdown": 2
|
||||||
|
|||||||
@@ -19,6 +19,8 @@ class DisplayController:
|
|||||||
self.weather = WeatherManager(self.config, self.display_manager)
|
self.weather = WeatherManager(self.config, self.display_manager)
|
||||||
self.current_display = 'clock'
|
self.current_display = 'clock'
|
||||||
self.last_switch = time.time()
|
self.last_switch = time.time()
|
||||||
|
self.hourly_index = 0
|
||||||
|
self.last_hourly_update = time.time()
|
||||||
logger.info("DisplayController initialized with display_manager: %s", id(self.display_manager))
|
logger.info("DisplayController initialized with display_manager: %s", id(self.display_manager))
|
||||||
|
|
||||||
def run(self):
|
def run(self):
|
||||||
@@ -27,26 +29,48 @@ class DisplayController:
|
|||||||
while True:
|
while True:
|
||||||
current_time = time.time()
|
current_time = time.time()
|
||||||
rotation_interval = self.config['display'].get('rotation_interval', 15)
|
rotation_interval = self.config['display'].get('rotation_interval', 15)
|
||||||
|
hourly_scroll_interval = 3 # Show each hour for 3 seconds
|
||||||
|
|
||||||
# Track if we're switching modes
|
# Track if we're switching modes
|
||||||
switching_modes = False
|
switching_modes = False
|
||||||
|
|
||||||
# Switch display if interval has passed
|
# Switch display if interval has passed
|
||||||
if current_time - self.last_switch > rotation_interval:
|
if current_time - self.last_switch > rotation_interval:
|
||||||
logger.info("Switching display from %s to %s",
|
# Cycle through: clock -> current weather -> hourly forecast -> daily forecast
|
||||||
self.current_display,
|
if self.current_display == 'clock':
|
||||||
'weather' if self.current_display == 'clock' else 'clock')
|
self.current_display = 'weather'
|
||||||
self.current_display = 'weather' if self.current_display == 'clock' else 'clock'
|
elif self.current_display == 'weather':
|
||||||
|
self.current_display = 'hourly'
|
||||||
|
self.hourly_index = 0
|
||||||
|
self.last_hourly_update = current_time
|
||||||
|
elif self.current_display == 'hourly':
|
||||||
|
self.current_display = 'daily'
|
||||||
|
else: # daily
|
||||||
|
self.current_display = 'clock'
|
||||||
|
|
||||||
|
logger.info("Switching display to: %s", self.current_display)
|
||||||
self.last_switch = current_time
|
self.last_switch = current_time
|
||||||
switching_modes = True
|
switching_modes = True
|
||||||
|
|
||||||
|
# Update hourly forecast index if needed
|
||||||
|
if self.current_display == 'hourly' and current_time - self.last_hourly_update > hourly_scroll_interval:
|
||||||
|
self.hourly_index = (self.hourly_index + 1) % 6 # We show 6 hours
|
||||||
|
self.last_hourly_update = current_time
|
||||||
|
switching_modes = True # Force clear for new hour
|
||||||
|
|
||||||
# Display current screen
|
# Display current screen
|
||||||
if self.current_display == 'clock':
|
if self.current_display == 'clock':
|
||||||
logger.debug("Updating clock display")
|
logger.debug("Updating clock display")
|
||||||
self.clock.display_time(force_clear=switching_modes)
|
self.clock.display_time(force_clear=switching_modes)
|
||||||
else:
|
elif self.current_display == 'weather':
|
||||||
logger.debug("Updating weather display")
|
logger.debug("Updating current weather display")
|
||||||
self.weather.display_weather(force_clear=switching_modes)
|
self.weather.display_weather(force_clear=switching_modes)
|
||||||
|
elif self.current_display == 'hourly':
|
||||||
|
logger.debug("Updating hourly forecast display")
|
||||||
|
self.weather.display_hourly_forecast(self.hourly_index, force_clear=switching_modes)
|
||||||
|
else: # daily
|
||||||
|
logger.debug("Updating daily forecast display")
|
||||||
|
self.weather.display_daily_forecast(force_clear=switching_modes)
|
||||||
|
|
||||||
# Sleep for 0.5 seconds since we only need to check for second changes
|
# Sleep for 0.5 seconds since we only need to check for second changes
|
||||||
time.sleep(0.5)
|
time.sleep(0.5)
|
||||||
|
|||||||
@@ -1,9 +1,29 @@
|
|||||||
import requests
|
import requests
|
||||||
import time
|
import time
|
||||||
from typing import Dict, Any
|
from datetime import datetime
|
||||||
|
from typing import Dict, Any, List
|
||||||
from PIL import Image, ImageDraw
|
from PIL import Image, ImageDraw
|
||||||
|
|
||||||
class WeatherManager:
|
class WeatherManager:
|
||||||
|
# Weather condition to emoji mapping
|
||||||
|
WEATHER_ICONS = {
|
||||||
|
'Clear': '☀️',
|
||||||
|
'Clouds': '☁️',
|
||||||
|
'Rain': '🌧️',
|
||||||
|
'Snow': '❄️',
|
||||||
|
'Thunderstorm': '⛈️',
|
||||||
|
'Drizzle': '🌦️',
|
||||||
|
'Mist': '🌫️',
|
||||||
|
'Fog': '🌫️',
|
||||||
|
'Haze': '🌫️',
|
||||||
|
'Smoke': '🌫️',
|
||||||
|
'Dust': '🌫️',
|
||||||
|
'Sand': '🌫️',
|
||||||
|
'Ash': '🌫️',
|
||||||
|
'Squall': '💨',
|
||||||
|
'Tornado': '🌪️'
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(self, config: Dict[str, Any], display_manager):
|
def __init__(self, config: Dict[str, Any], display_manager):
|
||||||
self.config = config
|
self.config = config
|
||||||
self.display_manager = display_manager
|
self.display_manager = display_manager
|
||||||
@@ -11,25 +31,93 @@ class WeatherManager:
|
|||||||
self.location = config.get('location', {})
|
self.location = config.get('location', {})
|
||||||
self.last_update = 0
|
self.last_update = 0
|
||||||
self.weather_data = None
|
self.weather_data = None
|
||||||
|
self.forecast_data = None
|
||||||
|
self.hourly_forecast = None
|
||||||
|
self.daily_forecast = None
|
||||||
|
|
||||||
def _fetch_weather(self) -> None:
|
def _fetch_weather(self) -> None:
|
||||||
"""Fetch weather data from OpenWeatherMap API."""
|
"""Fetch current weather and forecast data from OpenWeatherMap API."""
|
||||||
api_key = self.weather_config['api_key']
|
api_key = self.weather_config.get('api_key')
|
||||||
|
if not api_key:
|
||||||
|
print("No API key configured for weather")
|
||||||
|
return
|
||||||
|
|
||||||
city = self.location['city']
|
city = self.location['city']
|
||||||
state = self.location['state']
|
state = self.location['state']
|
||||||
country = self.location['country']
|
country = self.location['country']
|
||||||
units = self.weather_config.get('units', 'imperial')
|
units = self.weather_config.get('units', 'imperial')
|
||||||
|
|
||||||
url = f"http://api.openweathermap.org/data/2.5/weather?q={city},{state},{country}&appid={api_key}&units={units}"
|
# Get current weather
|
||||||
|
current_url = f"http://api.openweathermap.org/data/2.5/weather?q={city},{state},{country}&appid={api_key}&units={units}"
|
||||||
|
|
||||||
|
# Get forecast data (includes hourly and daily)
|
||||||
|
forecast_url = f"http://api.openweathermap.org/data/2.5/forecast?q={city},{state},{country}&appid={api_key}&units={units}"
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response = requests.get(url)
|
# Fetch current weather
|
||||||
|
response = requests.get(current_url)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
self.weather_data = response.json()
|
self.weather_data = response.json()
|
||||||
|
|
||||||
|
# Fetch forecast
|
||||||
|
response = requests.get(forecast_url)
|
||||||
|
response.raise_for_status()
|
||||||
|
self.forecast_data = response.json()
|
||||||
|
|
||||||
|
# Process forecast data
|
||||||
|
self._process_forecast_data()
|
||||||
|
|
||||||
self.last_update = time.time()
|
self.last_update = time.time()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"Error fetching weather data: {e}")
|
print(f"Error fetching weather data: {e}")
|
||||||
self.weather_data = None
|
self.weather_data = None
|
||||||
|
self.forecast_data = None
|
||||||
|
|
||||||
|
def _process_forecast_data(self) -> None:
|
||||||
|
"""Process the forecast data into hourly and daily forecasts."""
|
||||||
|
if not self.forecast_data:
|
||||||
|
return
|
||||||
|
|
||||||
|
# Process hourly forecast (next 6 hours)
|
||||||
|
self.hourly_forecast = []
|
||||||
|
for item in self.forecast_data['list'][:6]: # First 6 entries (3 hours each)
|
||||||
|
hour = datetime.fromtimestamp(item['dt']).strftime('%I%p')
|
||||||
|
temp = round(item['main']['temp'])
|
||||||
|
condition = item['weather'][0]['main']
|
||||||
|
icon = self.WEATHER_ICONS.get(condition, '❓')
|
||||||
|
self.hourly_forecast.append({
|
||||||
|
'hour': hour,
|
||||||
|
'temp': temp,
|
||||||
|
'condition': condition,
|
||||||
|
'icon': icon
|
||||||
|
})
|
||||||
|
|
||||||
|
# Process daily forecast (next 3 days)
|
||||||
|
daily_data = {}
|
||||||
|
for item in self.forecast_data['list']:
|
||||||
|
date = datetime.fromtimestamp(item['dt']).strftime('%Y-%m-%d')
|
||||||
|
if date not in daily_data:
|
||||||
|
daily_data[date] = {
|
||||||
|
'temps': [],
|
||||||
|
'conditions': []
|
||||||
|
}
|
||||||
|
daily_data[date]['temps'].append(item['main']['temp'])
|
||||||
|
daily_data[date]['conditions'].append(item['weather'][0]['main'])
|
||||||
|
|
||||||
|
# Calculate daily summaries
|
||||||
|
self.daily_forecast = []
|
||||||
|
for date, data in list(daily_data.items())[:3]: # First 3 days
|
||||||
|
avg_temp = round(sum(data['temps']) / len(data['temps']))
|
||||||
|
# Get most common condition for the day
|
||||||
|
condition = max(set(data['conditions']), key=data['conditions'].count)
|
||||||
|
icon = self.WEATHER_ICONS.get(condition, '❓')
|
||||||
|
display_date = datetime.strptime(date, '%Y-%m-%d').strftime('%a %d')
|
||||||
|
self.daily_forecast.append({
|
||||||
|
'date': display_date,
|
||||||
|
'temp': avg_temp,
|
||||||
|
'condition': condition,
|
||||||
|
'icon': icon
|
||||||
|
})
|
||||||
|
|
||||||
def get_weather(self) -> Dict[str, Any]:
|
def get_weather(self) -> Dict[str, Any]:
|
||||||
"""Get current weather data, fetching new data if needed."""
|
"""Get current weather data, fetching new data if needed."""
|
||||||
@@ -40,16 +128,51 @@ class WeatherManager:
|
|||||||
return self.weather_data
|
return self.weather_data
|
||||||
|
|
||||||
def display_weather(self, force_clear: bool = False) -> None:
|
def display_weather(self, force_clear: bool = False) -> None:
|
||||||
"""Display weather information on the LED matrix."""
|
"""Display current weather information on the LED matrix."""
|
||||||
weather_data = self.get_weather()
|
weather_data = self.get_weather()
|
||||||
if not weather_data:
|
if not weather_data:
|
||||||
return
|
return
|
||||||
|
|
||||||
temp = round(weather_data['main']['temp'])
|
temp = round(weather_data['main']['temp'])
|
||||||
condition = weather_data['weather'][0]['main']
|
condition = weather_data['weather'][0]['main']
|
||||||
|
icon = self.WEATHER_ICONS.get(condition, '❓')
|
||||||
|
|
||||||
# Format the display string with both temp and condition
|
# Format the display string with temp, icon, and condition
|
||||||
display_text = f"{temp}°F\n{condition}"
|
display_text = f"{temp}°F {icon}\n{condition}"
|
||||||
|
|
||||||
# Draw both lines at once using the multi-line support in draw_text
|
# Draw both lines at once using the multi-line support in draw_text
|
||||||
|
self.display_manager.draw_text(display_text, force_clear=force_clear)
|
||||||
|
|
||||||
|
def display_hourly_forecast(self, index: int = 0, force_clear: bool = False) -> None:
|
||||||
|
"""Display hourly forecast information, showing one time slot at a time."""
|
||||||
|
if not self.hourly_forecast:
|
||||||
|
self.get_weather() # This will also update forecasts
|
||||||
|
if not self.hourly_forecast:
|
||||||
|
return
|
||||||
|
|
||||||
|
# Get the forecast for the current index
|
||||||
|
forecast = self.hourly_forecast[index % len(self.hourly_forecast)]
|
||||||
|
|
||||||
|
# Format the display string
|
||||||
|
display_text = f"{forecast['hour']}\n{forecast['temp']}°F {forecast['icon']}"
|
||||||
|
|
||||||
|
# Draw the forecast
|
||||||
|
self.display_manager.draw_text(display_text, force_clear=force_clear)
|
||||||
|
|
||||||
|
def display_daily_forecast(self, force_clear: bool = False) -> None:
|
||||||
|
"""Display 3-day forecast information."""
|
||||||
|
if not self.daily_forecast:
|
||||||
|
self.get_weather() # This will also update forecasts
|
||||||
|
if not self.daily_forecast:
|
||||||
|
return
|
||||||
|
|
||||||
|
# Create a compact display of all three days
|
||||||
|
lines = []
|
||||||
|
for day in self.daily_forecast:
|
||||||
|
lines.append(f"{day['date']}: {day['temp']}°F {day['icon']}")
|
||||||
|
|
||||||
|
# Join all lines with newlines
|
||||||
|
display_text = "\n".join(lines)
|
||||||
|
|
||||||
|
# Draw the forecast
|
||||||
self.display_manager.draw_text(display_text, force_clear=force_clear)
|
self.display_manager.draw_text(display_text, force_clear=force_clear)
|
||||||
Reference in New Issue
Block a user