Upload climate_data.py

data/climate_data.py

@@ -21,9 +21,6 @@ from datetime import datetime, timedelta
 import re
 import logging
 import hashlib
-import pytz
-from typing import Dict, List
-from timezonefinder import TimezoneFinder
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -92,8 +89,9 @@ class ClimateLocation:
     _solar_cache: Dict[str, Dict[str, float]]  # Cache for solar angle calculations
     
     def __init__(self, epw_file: pd.DataFrame, typical_extreme_periods: Dict, ground_temperatures: Dict, albedo: float = 0.2, **kwargs):
+        """Initialize ClimateLocation with EPW file data and header information."""
         self.id = kwargs.get("id")
-        self.country = kwargs.get("country")  # Set country from kwargs
+        self.country = kwargs.get("country")
         self.state_province = kwargs.get("state_province", "N/A")
         self.city = kwargs.get("city")
         self.latitude = kwargs.get("latitude")
@@ -101,11 +99,10 @@ class ClimateLocation:
         self.elevation = kwargs.get("elevation")
         self.time_zone = kwargs.get("time_zone")
         self.albedo = albedo
-        self.hourly_data = epw_file.iloc[8:].to_dict('records')  # Assuming data starts at row 8
         self.typical_extreme_periods = typical_extreme_periods
         self.ground_temperatures = ground_temperatures
-        self.solstice_zenith_angles = {}
-        self.tz = pytz.FixedOffset(int(self.time_zone * 60))  # Fixed offset in minutes
+        self._solar_cache = {}
+        self.solstice_zenith_angles = {}  # Initialize empty, to be calculated later
         
         # Extract columns from EPW data
         months = pd.to_numeric(epw_file[1], errors='coerce').values
@@ -209,110 +206,78 @@ class ClimateLocation:
         return solar_time % 24
 
     def _calculate_solar_angles(self, timestamp: datetime) -> Dict[str, float]:
-        """
-        Compute solar zenith and azimuth angles for a single timestamp using pvlib and timezone-aware datetime.
+        """Compute solar zenith and azimuth angles using ASHRAE solar time."""
+        # Calculate solar time per ASHRAE
+        solar_time = self._calculate_solar_time(timestamp)
+        # Convert solar time back to a datetime object for pvlib
+        solar_hour = int(solar_time)
+        solar_minute = int((solar_time - solar_hour) * 60)
+        try:
+            solar_timestamp = datetime(
+                timestamp.year, timestamp.month, timestamp.day, solar_hour, solar_minute
+            )
+        except ValueError:
+            # Handle edge cases (e.g., solar time near midnight)
+            solar_timestamp = timestamp + timedelta(hours=solar_time - timestamp.hour)
         
-        Args:
-            timestamp (datetime): Datetime object (naive, will be localized to the location's timezone).
+        # Generate cache key including solar time for precision
+        cache_key = hashlib.md5(
+            f"{solar_timestamp.strftime('%Y-%m-%d_%H:%M')}_{self.latitude}_{self.longitude}_{self.time_zone}".encode()
+        ).hexdigest()
+        
+        if cache_key in self._solar_cache:
+            return self._solar_cache[cache_key]
         
-        Returns:
-            Dict[str, float]: Dictionary with 'zenith' and 'azimuth' angles in degrees.
-        """
-        # Make the timestamp timezone-aware
-        local_time = self.tz.localize(timestamp)
-
-        # Get solar position
         solpos = pvlib.solarposition.get_solarposition(
-            time=pd.DatetimeIndex([local_time]),
+            time=solar_timestamp,
             latitude=self.latitude,
             longitude=self.longitude,
             altitude=self.elevation,
-            method='nrel_numpy'  # High-accuracy method for ASHRAE compliance
+            method='nrel_numpy',  # Consistent method
+            delta_t=0  # Disable DST adjustments
         )
-
-        return {
+        result = {
             'zenith': float(solpos['zenith'].iloc[0]),
             'azimuth': float(solpos['azimuth'].iloc[0])
         }
+        self._solar_cache[cache_key] = result
+        return result
 
-    def _calculate_solar_angles_batch(self, timestamps: List[datetime]) -> List[Dict[str, float]]:
-        """
-        Compute solar zenith and azimuth angles for multiple timestamps (optimized for large datasets).
-        
-        Args:
-            timestamps (List[datetime]): List of naive datetime objects to process.
-        
-        Returns:
-            List[Dict[str, float]]: List of dictionaries with 'zenith' and 'azimuth' angles in degrees.
-        """
-        # Localize all timestamps
-        local_times = [self.tz.localize(ts) for ts in timestamps]
-
-        # Get solar positions in a single vectorized call
-        solpos = pvlib.solarposition.get_solarposition(
-            time=pd.DatetimeIndex(local_times),
-            latitude=self.latitude,
-            longitude=self.longitude,
-            altitude=self.elevation,
-            method='nrel_numpy'
-        )
-
-        return [
-            {'zenith': float(solpos['zenith'].iloc[i]), 'azimuth': float(solpos['azimuth'].iloc[i])}
-            for i in range(len(timestamps))
-        ]
-        
     def calculate_ground_reflection(self):
-        """Calculate solar time, angles, and ground-reflected radiation for hourly data."""
+        """Calculate solar time, angles, ground-reflected radiation, and solstice zenith angles for hourly data."""
         zenith_angles = []
         for i, record in enumerate(self.hourly_data):
+            # Create timestamp (adjust hour from 1-24 to 0-23)
             try:
-                # Set timestamp to the end of the hour
-                if record['hour'] == 24:
-                    # Handle hour=24 as 00:00 of the next day
-                    if record['month'] == 12 and record['day'] == 31:
-                        timestamp = datetime(2026, 1, 1, 0)  # Next year if end of December
-                    else:
-                        timestamp = datetime(2025, record['month'], record['day']) + timedelta(days=1)
-                        timestamp = timestamp.replace(hour=0)
-                else:
-                    timestamp = datetime(2025, record['month'], record['day'], record['hour'])
+                timestamp = datetime(2025, record['month'], record['day'], record['hour'] - 1)
             except ValueError:
-                logger.warning(f"Invalid date at index {i}: {record['month']}/{record['day']} {record['hour']}")
+                logger.warning(f"Invalid date in hourly data at index {i}: {record['month']}/{record['day']} {record['hour']}")
                 continue
-
-            # Localize timestamp to fixed UTC+10
-            local_time = self.tz.localize(timestamp)
-
-            # Calculate solar angles using pvlib
-            solpos = pvlib.solarposition.get_solarposition(
-                time=pd.DatetimeIndex([local_time]),
-                latitude=self.latitude,
-                longitude=self.longitude,
-                altitude=self.elevation,
-                method='nrel_numpy'
-            )
-            record['solar_zenith'] = round(float(solpos['zenith'].iloc[0]), 1)
-            record['solar_azimuth'] = round(float(solpos['azimuth'].iloc[0]), 1)
-            zenith_angles.append(record['solar_zenith'])
-
+            
             # Calculate solar time
-            solar_time = pvlib.solarposition.solar_time(
-                time=pd.DatetimeIndex([local_time]),
-                longitude=self.longitude
-            )
-            record['solar_time'] = round((solar_time.hour + solar_time.minute / 60.0) % 24, 2)
-
+            record['solar_time'] = round(self._calculate_solar_time(timestamp), 2)
+            
+            # Calculate solar angles
+            solar_angles = self._calculate_solar_angles(timestamp)
+            record['solar_zenith'] = round(solar_angles['zenith'], 1)
+            record['solar_azimuth'] = round(solar_angles['azimuth'], 1)
+            zenith_angles.append(record['solar_zenith'])
+            
             # Calculate ground-reflected radiation using Perez model
             dni = record['direct_normal_irradiance']
             dhi = record['diffuse_horizontal_irradiance']
             if dhi == 0:
+                # Skip Perez model if dhi is zero to avoid division by zero
                 record['ground_reflected_radiation'] = 0.0
             else:
-                dni_extra = pvlib.irradiance.get_extra_radiation(local_time)
+                # Create a single timestamp index for pvlib
+                index = pd.DatetimeIndex([timestamp])
+                # Calculate extraterrestrial DNI for Perez model
+                dni_extra = pvlib.irradiance.get_extra_radiation(timestamp)
+                # Calculate total irradiance on a horizontal surface
                 total_irradiance = pvlib.irradiance.get_total_irradiance(
-                    surface_tilt=0,
-                    surface_azimuth=0,
+                    surface_tilt=0,  # Horizontal surface
+                    surface_azimuth=0,  # Irrelevant for horizontal
                     solar_zenith=record['solar_zenith'],
                     solar_azimuth=record['solar_azimuth'],
                     dni=dni,
@@ -322,19 +287,21 @@ class ClimateLocation:
                     albedo=self.albedo,
                     model='perez'
                 )
+                # Handle both Series and scalar outputs for poa_ground_diffuse
                 ground_reflected = total_irradiance['poa_ground_diffuse']
                 if isinstance(ground_reflected, pd.Series):
                     ground_reflected = ground_reflected.iloc[0]
+                # Ensure the value is a float and handle potential NaN
                 record['ground_reflected_radiation'] = round(max(0.0, float(ground_reflected)), 1)
-
-        # Calculate solstice zenith angles
+        
+        # Calculate solstice zenith angles based on max/min zenith angles
         if zenith_angles:
             self.solstice_zenith_angles = {
                 'summer': round(min(zenith_angles), 1),  # Lowest zenith angle (highest sun)
                 'winter': round(max(zenith_angles), 1)   # Highest zenith angle (lowest sun)
             }
         else:
-            logger.warning("No valid zenith angles calculated.")
+            logger.warning("No valid zenith angles calculated, solstice_zenith_angles remain empty.")
             self.solstice_zenith_angles = {'summer': None, 'winter': None}
 
     def to_dict(self) -> Dict[str, Any]: