Compare commits

...

3 Commits

8 changed files with 210 additions and 27 deletions

View File

@ -0,0 +1,8 @@
---
description: ByteRover MCP Custom Instructions
alwaysApply: true
---
# important
always use search-memories tool to get the related context before any tasks
always use create-memories to store all the critical informations after sucessfull tasks

View File

@ -148,8 +148,35 @@ UV_THRESHOLD_HIGH=10
RISK_THRESHOLD_SHOES=6
ROLLING_WINDOW_HOURS=2
SURFACE_RECOVERY_HOURS=2
# Surface recovery enhancements (Optional)
SURFACE_RECOVERY_TEMP_THRESHOLD=90 # Temperature threshold to start recovery period (°F)
SURFACE_TYPE=asphalt # Options: asphalt, concrete, mixed, grass
SURFACE_MAX_RECOVERY_SCORE=2.0 # Maximum recovery score reduction with graduated recovery
ENABLE_GRADUATED_RECOVERY=true # Enable graduated recovery (more cooling time = more benefit)
ENABLE_TIME_OF_DAY_FACTOR=true # Apply day/night cooling adjustments
```
### Surface Recovery Feature
The application includes an enhanced surface recovery model that accounts for cooling of surfaces after peak temperatures:
- **Basic Recovery**: Surfaces begin to cool after being exposed to high temperatures (above `SURFACE_RECOVERY_TEMP_THRESHOLD`). After `SURFACE_RECOVERY_HOURS`, a recovery credit reduces the risk score.
- **Graduated Recovery**: When enabled, provides a progressively stronger recovery benefit as more cooling time passes, up to `SURFACE_MAX_RECOVERY_SCORE`.
- **Surface Type**: Different surfaces cool at different rates:
- Asphalt: Slowest cooling (coefficient 0.7)
- Concrete: Moderate cooling (coefficient 0.85)
- Mixed: Standard cooling (coefficient 1.0)
- Grass: Fastest cooling (coefficient 1.5)
- **Time-of-Day**: When enabled, night hours (7pm-6am) provide 30% faster cooling than daylight hours.
- **Sun Exposure**: The algorithm considers sun exposure during the recovery period, which can slow cooling by up to 30%.
These factors combine to provide a more accurate assessment of surface temperatures throughout the day and the resulting paw burn risk.
## Output Examples
### Summary Output

View File

@ -23,6 +23,16 @@ class RiskConfig:
rolling_window_hours: int = 2
surface_recovery_hours: int = 2
# Surface recovery enhancement parameters
surface_recovery_temp_threshold: float = 90.0
surface_type: str = "asphalt" # Options: asphalt, concrete, mixed, grass
surface_max_recovery_score: float = 2.0
enable_graduated_recovery: bool = True
enable_time_of_day_factor: bool = True
# Display preferences
use_24hr_time: bool = False # False for 12hr time format, True for 24hr
@classmethod
def from_env(cls) -> 'RiskConfig':
"""Create configuration from environment variables."""
@ -35,7 +45,15 @@ class RiskConfig:
uv_threshold_high=float(os.getenv('UV_THRESHOLD_HIGH', 10)),
risk_threshold_shoes=float(os.getenv('RISK_THRESHOLD_SHOES', 6)),
rolling_window_hours=int(os.getenv('ROLLING_WINDOW_HOURS', 2)),
surface_recovery_hours=int(os.getenv('SURFACE_RECOVERY_HOURS', 2))
surface_recovery_hours=int(os.getenv('SURFACE_RECOVERY_HOURS', 2)),
# Surface recovery enhancement parameters
surface_recovery_temp_threshold=float(os.getenv('SURFACE_RECOVERY_TEMP_THRESHOLD', 90)),
surface_type=os.getenv('SURFACE_TYPE', 'asphalt'),
surface_max_recovery_score=float(os.getenv('SURFACE_MAX_RECOVERY_SCORE', 2.0)),
enable_graduated_recovery=os.getenv('ENABLE_GRADUATED_RECOVERY', 'true').lower() == 'true',
enable_time_of_day_factor=os.getenv('ENABLE_TIME_OF_DAY_FACTOR', 'true').lower() == 'true',
# Display preferences
use_24hr_time=os.getenv('USE_24HR_TIME', 'false').lower() == 'true'
)
@dataclass

17
constants.py Normal file
View File

@ -0,0 +1,17 @@
"""Constants used throughout the application."""
# Surface cooling coefficients (slower to faster cooling)
SURFACE_COOLING_COEFFICIENTS = {
'asphalt': 0.7, # Slower cooling
'concrete': 0.85,
'mixed': 1.0, # Normal cooling
'grass': 1.5 # Faster cooling
}
# Night hours definition (7pm to 6am)
NIGHT_START_HOUR = 19 # 7pm
NIGHT_END_HOUR = 6 # 6am
# Time of day cooling multipliers
NIGHT_COOLING_MULTIPLIER = 1.3 # Night cools 30% faster
DAY_COOLING_MULTIPLIER = 1.0 # Standard cooling during day

View File

@ -13,16 +13,26 @@ DEFAULT_LOCATION=New York
DATABASE_PATH=paw_risk.db
# Temperature thresholds in Fahrenheit (Optional)
TEMP_THRESHOLD_LOW=80
TEMP_THRESHOLD_MED=90
TEMP_THRESHOLD_HIGH=100
TEMP_THRESHOLD_LOW=75
TEMP_THRESHOLD_MED=85
TEMP_THRESHOLD_HIGH=95
# UV Index thresholds (Optional)
UV_THRESHOLD_LOW=6
UV_THRESHOLD_MED=8
UV_THRESHOLD_HIGH=10
UV_THRESHOLD_LOW=5
UV_THRESHOLD_MED=7.5
UV_THRESHOLD_HIGH=9
# Risk assessment parameters (Optional)
RISK_THRESHOLD_SHOES=6
RISK_THRESHOLD_SHOES=5
ROLLING_WINDOW_HOURS=2
SURFACE_RECOVERY_HOURS=2
SURFACE_RECOVERY_HOURS=2
# Surface recovery enhancements (Optional)
SURFACE_RECOVERY_TEMP_THRESHOLD=90
SURFACE_TYPE=asphalt
SURFACE_MAX_RECOVERY_SCORE=2.0
ENABLE_GRADUATED_RECOVERY=true
ENABLE_TIME_OF_DAY_FACTOR=true
# Display preferences (Optional)
USE_24HR_TIME=false

14
main.py
View File

@ -71,6 +71,13 @@ class PawRiskApp:
logger.error(f"Error during analysis: {e}")
return {"error": str(e)}
def format_time(self, dt: datetime) -> str:
"""Format time based on user's preference."""
if self.config.risk_config.use_24hr_time:
return dt.strftime("%H:%M")
else:
return dt.strftime("%I:%M %p")
def print_summary(self, analysis_result: dict):
"""Print a formatted summary of the analysis."""
if "error" in analysis_result:
@ -117,11 +124,11 @@ class PawRiskApp:
print("\n🕐 HOURLY BREAKDOWN:")
print("-" * 80)
print(f"{'Time':>6} {'Temp':>6} {'UV':>4} {'Condition':>12} {'Risk':>6} {'Shoes':>7}")
print(f"{'Time':>8} {'Temp':>6} {'UV':>4} {'Condition':>12} {'Risk':>6} {'Shoes':>7}")
print("-" * 80)
for weather, risk in zip(weather_hours, risk_scores):
time_str = weather.datetime.strftime("%H:%M")
time_str = self.format_time(weather.datetime)
temp_str = f"{weather.temperature_f:.0f}°F"
uv_str = f"{weather.uv_index:.1f}" if weather.uv_index else "N/A"
condition_short = weather.condition[:12]
@ -129,7 +136,7 @@ class PawRiskApp:
shoes_str = "YES" if risk.recommend_shoes else "no"
shoes_color = "⚠️ " if risk.recommend_shoes else ""
print(f"{time_str:>6} {temp_str:>6} {uv_str:>4} {condition_short:>12} "
print(f"{time_str:>8} {temp_str:>6} {uv_str:>4} {condition_short:>12} "
f"{risk_str:>6} {shoes_color}{shoes_str:>5}")
def create_plots(self, analysis_result: dict, save_plots: bool = False):
@ -207,6 +214,7 @@ def main():
print(f"Default Location: {config.default_location}")
print(f"Database Path: {config.database_path}")
print(f"Risk Threshold: {config.risk_config.risk_threshold_shoes}")
print(f"Time Format: {'24-hour' if config.risk_config.use_24hr_time else '12-hour'}")
return
except Exception as e:
print(f"❌ Configuration error: {e}")

View File

@ -10,6 +10,7 @@ import warnings
import os
import shutil
from models import WeatherHour, RiskScore
from config import RiskConfig, get_config
# Suppress matplotlib UserWarnings and macOS GUI warnings
warnings.filterwarnings('ignore', category=UserWarning, module='matplotlib')
@ -18,16 +19,30 @@ warnings.filterwarnings('ignore', message='.*NSSavePanel.*')
class RiskPlotter:
"""Handles plotting and visualization of risk data."""
def __init__(self, figure_size: Tuple[int, int] = (12, 8)):
def __init__(self, figure_size: Tuple[int, int] = (12, 8), use_24hr_time: Optional[bool] = None):
self.figure_size = figure_size
self.plots_dir = "plots"
self._plots_dir_setup = False
# Get time format preference from config if not specified
if use_24hr_time is None:
config = get_config().risk_config
self.use_24hr_time = getattr(config, 'use_24hr_time', False)
else:
self.use_24hr_time = use_24hr_time
# Set up matplotlib style and suppress warnings
plt.style.use('default')
plt.rcParams['figure.figsize'] = figure_size
plt.rcParams['font.size'] = 10
def get_time_formatter(self):
"""Get the appropriate time formatter based on config."""
if self.use_24hr_time:
return mdates.DateFormatter('%H:%M')
else:
return mdates.DateFormatter('%I:%M %p')
def _setup_plots_directory(self):
"""Create and clear plots directory."""
if os.path.exists(self.plots_dir):
@ -116,8 +131,9 @@ class RiskPlotter:
ax3.grid(True, alpha=0.3)
# Format x-axis
time_formatter = self.get_time_formatter()
for ax in [ax1, ax2, ax3]:
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M'))
ax.xaxis.set_major_formatter(time_formatter)
ax.xaxis.set_major_locator(mdates.HourLocator(interval=2))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
@ -166,7 +182,8 @@ class RiskPlotter:
ax.grid(True, alpha=0.3)
# Format x-axis
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M'))
time_formatter = self.get_time_formatter()
ax.xaxis.set_major_formatter(time_formatter)
ax.xaxis.set_major_locator(mdates.HourLocator(interval=2))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
@ -334,8 +351,9 @@ class RiskPlotter:
ax4.set_title('Weather Conditions')
# Format x-axis for all subplots
time_formatter = self.get_time_formatter()
for ax in [ax1, ax4]:
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M'))
ax.xaxis.set_major_formatter(time_formatter)
ax.xaxis.set_major_locator(mdates.HourLocator(interval=2))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)

View File

@ -5,6 +5,8 @@ from datetime import datetime, timedelta
from typing import List, Optional, Tuple
from models import WeatherHour, RiskScore
from config import RiskConfig, get_config
from constants import SURFACE_COOLING_COEFFICIENTS, NIGHT_START_HOUR, NIGHT_END_HOUR
from constants import NIGHT_COOLING_MULTIPLIER, DAY_COOLING_MULTIPLIER
logger = logging.getLogger(__name__)
@ -13,6 +15,21 @@ class RiskCalculator:
def __init__(self, config: Optional[RiskConfig] = None):
self.config = config or get_config().risk_config
# Validate surface type is supported
if self.config.surface_type.lower() not in SURFACE_COOLING_COEFFICIENTS:
logger.warning(f"Surface type '{self.config.surface_type}' not recognized, using 'mixed' instead")
self.config.surface_type = "mixed"
# Log enhanced recovery settings
if self.config.enable_graduated_recovery:
logger.info(f"Using graduated recovery scoring with max score of {self.config.surface_max_recovery_score}")
if self.config.enable_time_of_day_factor:
logger.info(f"Time-of-day cooling factor enabled (night: {NIGHT_COOLING_MULTIPLIER}x, day: {DAY_COOLING_MULTIPLIER}x)")
logger.info(f"Surface recovery config: threshold={self.config.surface_recovery_temp_threshold}°F, "
f"hours={self.config.surface_recovery_hours}, surface={self.config.surface_type}")
def calculate_temperature_score(self, temperature_f: float) -> float:
"""Calculate risk score based on air temperature."""
@ -78,25 +95,78 @@ class RiskCalculator:
def calculate_surface_recovery_score(self, weather_hours: List[WeatherHour],
current_index: int) -> float:
"""Calculate surface recovery score (time since last peak temperature)."""
"""Calculate surface recovery score with enhanced logic."""
if current_index < 2:
return 0.0
# Look back to find the last time temperature was ≥90°F
# Look back to find the last time temperature was above the threshold
hours_since_peak = 0
sun_exposure_hours = 0
peak_temp = 0
current_hour = weather_hours[current_index].datetime
for i in range(current_index - 1, -1, -1):
hours_since_peak += 1
if weather_hours[i].temperature_f >= 90.0:
# Check if temperature was above recovery threshold
if weather_hours[i].temperature_f >= self.config.surface_recovery_temp_threshold:
peak_temp = weather_hours[i].temperature_f
break
# Count sun exposure during recovery period
condition = weather_hours[i].condition.lower()
if 'sunny' in condition or 'clear' in condition:
sun_exposure_hours += 1
else:
# No peak found in available data
hours_since_peak = current_index + 1
# Give recovery credit if it's been >2 hours since last 90°F reading
if hours_since_peak > self.config.surface_recovery_hours:
return -1.0
else:
# No recovery needed if no peak was found
if peak_temp == 0:
return 0.0
# Apply surface type coefficient
cooling_coefficient = SURFACE_COOLING_COEFFICIENTS.get(
self.config.surface_type.lower(), 1.0)
# Apply time-of-day factor if enabled
if self.config.enable_time_of_day_factor:
hour_of_day = current_hour.hour
time_multiplier = (NIGHT_COOLING_MULTIPLIER
if hour_of_day >= NIGHT_START_HOUR or hour_of_day < NIGHT_END_HOUR
else DAY_COOLING_MULTIPLIER)
else:
time_multiplier = 1.0
# Calculate sun exposure percentage during recovery
sun_percentage = sun_exposure_hours / hours_since_peak if hours_since_peak > 0 else 0
# Sun slows cooling (reduce coefficient by up to 30%)
sun_factor = 1.0 - (sun_percentage * 0.3)
# Final adjusted hours since peak temperature
adjusted_hours = hours_since_peak * cooling_coefficient * time_multiplier * sun_factor
# Calculate recovery score
if adjusted_hours <= self.config.surface_recovery_hours:
# No recovery credit yet
return 0.0
elif self.config.enable_graduated_recovery:
# Graduated recovery (more hours = more recovery credit)
# Normalize to range 0.0 to max_recovery_score
hours_over = adjusted_hours - self.config.surface_recovery_hours
max_additional_hours = self.config.surface_recovery_hours # Full credit after double the recovery time
factor = min(1.0, hours_over / max_additional_hours)
recovery_score = -factor * self.config.surface_max_recovery_score
logger.debug(f"Graduated recovery: {hours_since_peak} hrs since {peak_temp}°F peak, "
f"adjusted to {adjusted_hours:.1f} hrs, score: {recovery_score}")
return recovery_score
else:
# Original binary approach
return -1.0
def interpolate_missing_uv(self, weather_hours: List[WeatherHour]) -> List[WeatherHour]:
"""Interpolate missing UV values using nearby hours."""
@ -233,6 +303,13 @@ class RiskCalculator:
return blocks
def format_time(self, dt: datetime) -> str:
"""Format time based on user's preference."""
if hasattr(self.config, 'use_24hr_time') and self.config.use_24hr_time:
return dt.strftime("%H:%M")
else:
return dt.strftime("%I:%M %p")
def generate_recommendations(self, risk_scores: List[RiskScore]) -> dict:
"""Generate comprehensive recommendations based on risk scores."""
if not risk_scores:
@ -256,13 +333,13 @@ class RiskCalculator:
"high_risk_hours": high_risk_hours,
"max_risk_score": round(max_score, 1),
"average_risk_score": round(avg_score, 1),
"peak_risk_time": peak_score.datetime.strftime("%H:%M"),
"peak_risk_time": self.format_time(peak_score.datetime),
"continuous_risk_blocks": len(risk_blocks)
},
"risk_periods": [
{
"start": start.strftime("%H:%M"),
"end": end.strftime("%H:%M"),
"start": self.format_time(start),
"end": self.format_time(end),
"duration_hours": round((end - start).total_seconds() / 3600, 1)
}
for start, end in risk_blocks