""" ASHRAE 169 climate data module for HVAC Load Calculator. Extracts climate data from EPW files and provides visualizations inspired by Climate Consultant. Author: Dr Majed Abuseif Date: May 2025 Version: 2.1.0 """ from typing import Dict, List, Any, Optional import pandas as pd import numpy as np import os import json from dataclasses import dataclass import streamlit as st import plotly.graph_objects as go from io import StringIO import pvlib from datetime import datetime, timedelta # Define paths DATA_DIR = os.path.dirname(os.path.abspath(__file__)) @dataclass class ClimateLocation: """Class representing a climate location with ASHRAE 169 data derived from EPW files.""" id: str country: str state_province: str city: str latitude: float longitude: float elevation: float # meters climate_zone: str heating_degree_days: float # base 18°C cooling_degree_days: float # base 18°C winter_design_temp: float # 99.6% heating design temperature (°C) summer_design_temp_db: float # 0.4% cooling design dry-bulb temperature (°C) summer_design_temp_wb: float # 0.4% cooling design wet-bulb temperature (°C) summer_daily_range: float # Mean daily temperature range in summer (°C) wind_speed: float # Mean wind speed (m/s) pressure: float # Atmospheric pressure (Pa) hourly_data: List[Dict] # Hourly data for integration with main.py def __init__(self, epw_file: pd.DataFrame, **kwargs): """Initialize ClimateLocation with EPW file data.""" self.id = kwargs.get("id") self.country = kwargs.get("country") self.state_province = kwargs.get("state_province", "N/A") self.city = kwargs.get("city") self.latitude = kwargs.get("latitude") self.longitude = kwargs.get("longitude") self.elevation = kwargs.get("elevation") months = pd.to_numeric(epw_file[1], errors='coerce').values dry_bulb = pd.to_numeric(epw_file[6], errors='coerce').values humidity = pd.to_numeric(epw_file[8], errors='coerce').values pressure = pd.to_numeric(epw_file[9], errors='coerce').values wind_speed = pd.to_numeric(epw_file[21], errors='coerce').values wind_direction = pd.to_numeric(epw_file[20], errors='coerce').values global_radiation = pd.to_numeric(epw_file[13], errors='coerce').values wet_bulb = ClimateData.calculate_wet_bulb(dry_bulb, humidity) self.winter_design_temp = round(np.nanpercentile(dry_bulb, 0.4), 1) self.summer_design_temp_db = round(np.nanpercentile(dry_bulb, 99.6), 1) self.summer_design_temp_wb = round(np.nanpercentile(wet_bulb, 99.6), 1) daily_temps = np.nanmean(dry_bulb.reshape(-1, 24), axis=1) self.heating_degree_days = round(np.nansum(np.maximum(18 - daily_temps, 0))) self.cooling_degree_days = round(np.nansum(np.maximum(daily_temps - 18, 0))) summer_mask = (months >= 6) & (months <= 8) summer_temps = dry_bulb[summer_mask].reshape(-1, 24) self.summer_daily_range = round(np.nanmean(np.nanmax(summer_temps, axis=1) - np.nanmin(summer_temps, axis=1)), 1) self.wind_speed = round(np.nanmean(wind_speed), 1) self.pressure = round(np.nanmean(pressure), 1) self.climate_zone = ClimateData.assign_climate_zone(self.heating_degree_days, self.cooling_degree_days, np.nanmean(humidity)) # Store hourly data for main.py integration self.hourly_data = [ { "month": int(months[i]), "hour": i % 24, "dry_bulb": float(dry_bulb[i]), "relative_humidity": float(humidity[i]), "global_horizontal_radiation": float(global_radiation[i]), "wind_speed": float(wind_speed[i]), "wind_direction": float(wind_direction[i]) } for i in range(len(months)) if not any(np.isnan([months[i], dry_bulb[i], humidity[i], global_radiation[i], wind_speed[i], wind_direction[i]])) ] def to_dict(self) -> Dict[str, Any]: """Convert the climate location to a dictionary.""" return { "id": self.id, "country": self.country, "state_province": self.state_province, "city": self.city, "latitude": self.latitude, "longitude": self.longitude, "elevation": self.elevation, "climate_zone": self.climate_zone, "heating_degree_days": self.heating_degree_days, "cooling_degree_days": self.cooling_degree_days, "winter_design_temp": self.winter_design_temp, "summer_design_temp_db": self.summer_design_temp_db, "summer_design_temp_wb": self.summer_design_temp_wb, "summer_daily_range": self.summer_daily_range, "wind_speed": self.wind_speed, "pressure": self.pressure, "hourly_data": self.hourly_data } class ClimateData: """Class for managing ASHRAE 169 climate data from EPW files.""" def __init__(self): """Initialize climate data.""" self.locations = {} self.countries = [] self.country_states = {} def add_location(self, location: ClimateLocation): """Add a new location to the dictionary.""" self.locations[location.id] = location self.countries = sorted(list(set(loc.country for loc in self.locations.values()))) self.country_states = self._group_locations_by_country_state() def _group_locations_by_country_state(self) -> Dict[str, Dict[str, List[str]]]: """Group locations by country and state/province.""" result = {} for loc in self.locations.values(): if loc.country not in result: result[loc.country] = {} if loc.state_province not in result[loc.country]: result[loc.country][loc.state_province] = [] result[loc.country][loc.state_province].append(loc.city) for country in result: for state in result[country]: result[country][state] = sorted(result[country][state]) return result def get_location_by_id(self, location_id: str, session_state: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Retrieve climate data by ID from session state or locations.""" if "climate_data" in session_state and session_state["climate_data"].get("id") == location_id: return session_state["climate_data"] if location_id in self.locations: return self.locations[location_id].to_dict() return None @staticmethod def validate_climate_data(data: Dict[str, Any]) -> bool: """Validate climate data for required fields and ranges.""" required_fields = [ "id", "country", "city", "latitude", "longitude", "elevation", "climate_zone", "heating_degree_days", "cooling_degree_days", "winter_design_temp", "summer_design_temp_db", "summer_design_temp_wb", "summer_daily_range", "wind_speed", "pressure", "hourly_data" ] for field in required_fields: if field not in data: return False if not (-90 <= data["latitude"] <= 90 and -180 <= data["longitude"] <= 180): return False if data["elevation"] < 0: return False if data["climate_zone"] not in ["0A", "0B", "1A", "1B", "2A", "2B", "3A", "3B", "3C", "4A", "4B", "4C", "5A", "5B", "5C", "6A", "6B", "7", "8"]: return False if not (data["heating_degree_days"] >= 0 and data["cooling_degree_days"] >= 0): return False if not (-50 <= data["winter_design_temp"] <= 20): return False if not (0 <= data["summer_design_temp_db"] <= 50 and 0 <= data["summer_design_temp_wb"] <= 40): return False if data["summer_daily_range"] < 0: return False if not (0 <= data["wind_speed"] <= 20): return False if not (50000 <= data["pressure"] <= 120000): return False if not data["hourly_data"] or len(data["hourly_data"]) != 8760: return False for record in data["hourly_data"]: if not (1 <= record["month"] <= 12 and 0 <= record["hour"] <= 23): return False if not (-50 <= record["dry_bulb"] <= 50): return False if not (0 <= record["relative_humidity"] <= 100): return False if not (0 <= record["global_horizontal_radiation"] <= 1200): return False if not (0 <= record["wind_speed"] <= 20): return False if not (0 <= record["wind_direction"] <= 360): return False return True @staticmethod def calculate_wet_bulb(dry_bulb: np.ndarray, relative_humidity: np.ndarray) -> np.ndarray: """Calculate Wet Bulb Temperature using Stull (2011) approximation.""" db = np.array(dry_bulb, dtype=float) rh = np.array(relative_humidity, dtype=float) term1 = db * np.arctan(0.151977 * (rh + 8.313659)**0.5) term2 = np.arctan(db + rh) term3 = np.arctan(rh - 1.676331) term4 = 0.00391838 * rh**1.5 * np.arctan(0.023101 * rh) term5 = -4.686035 wet_bulb = term1 + term2 - term3 + term4 + term5 invalid_mask = (rh < 5) | (rh > 99) | (db < -20) | (db > 50) | np.isnan(db) | np.isnan(rh) wet_bulb[invalid_mask] = np.nan return wet_bulb def display_climate_input(self, session_state: Dict[str, Any]): """Display Streamlit interface for EPW upload and visualizations.""" st.title("Climate Data Analysis") uploaded_file = st.file_uploader("Upload EPW File", type=["epw"]) # Initialize location and epw_data for display location = None epw_data = None if uploaded_file: try: # Process new EPW file epw_content = uploaded_file.read().decode("utf-8") epw_lines = epw_content.splitlines() header = next(line for line in epw_lines if line.startswith("LOCATION")) header_parts = header.split(",") city = header_parts[1].strip() state_province = header_parts[2].strip() or "N/A" country = header_parts[3].strip() latitude = float(header_parts[6]) longitude = float(header_parts[7]) elevation = float(header_parts[8]) data_start_idx = next(i for i, line in enumerate(epw_lines) if line.startswith("DATA PERIODS")) + 1 epw_data = pd.read_csv(StringIO("\n".join(epw_lines[data_start_idx:])), header=None, dtype=str) if len(epw_data) != 8760: raise ValueError(f"EPW file has {len(epw_data)} records, expected 8760.") if len(epw_data.columns) != 35: raise ValueError(f"EPW file has {len(epw_data.columns)} columns, expected 35.") for col in [1, 6, 8, 9, 13, 20, 21]: epw_data[col] = pd.to_numeric(epw_data[col], errors='coerce') if epw_data[col].isna().all(): raise ValueError(f"Column {col} contains only non-numeric or missing data.") location = ClimateLocation( epw_file=epw_data, id=f"{country[:1].upper()}{city[:3].upper()}", country=country, state_province=state_province, city=city, latitude=latitude, longitude=longitude, elevation=elevation ) self.add_location(location) climate_data_dict = location.to_dict() if not self.validate_climate_data(climate_data_dict): raise ValueError("Invalid climate data extracted from EPW file.") session_state["climate_data"] = climate_data_dict st.success("Climate data extracted from EPW file!") except Exception as e: st.error(f"Error processing EPW file: {str(e)}. Ensure it has 8760 hourly records and correct format.") elif "climate_data" in session_state and self.validate_climate_data(session_state["climate_data"]): # Reconstruct from session_state climate_data_dict = session_state["climate_data"] # Rebuild epw_data from hourly_data hourly_data = climate_data_dict["hourly_data"] epw_data = pd.DataFrame({ 1: [d["month"] for d in hourly_data], # Month 6: [d["dry_bulb"] for d in hourly_data], # Dry-bulb temperature 8: [d["relative_humidity"] for d in hourly_data], # Relative humidity 9: [climate_data_dict["pressure"]] * len(hourly_data), # Pressure (mean value) 13: [d["global_horizontal_radiation"] for d in hourly_data], # Global horizontal radiation 20: [d["wind_direction"] for d in hourly_data], # Wind direction 21: [d["wind_speed"] for d in hourly_data], # Wind speed }) # Create ClimateLocation with reconstructed epw_data location = ClimateLocation( epw_file=epw_data, id=climate_data_dict["id"], country=climate_data_dict["country"], state_province=climate_data_dict["state_province"], city=climate_data_dict["city"], latitude=climate_data_dict["latitude"], longitude=climate_data_dict["longitude"], elevation=climate_data_dict["elevation"] ) # Override hourly_data to ensure consistency location.hourly_data = climate_data_dict["hourly_data"] self.add_location(location) st.info("Displaying previously extracted climate data.") # Display tabs if location and epw_data are available if location and epw_data is not None: tab1, tab2, tab3, tab4, tab5 = st.tabs([ "General Information", "Psychrometric Chart", "Sun Shading Chart", "Temperature Range", "Wind Rose" ]) with tab1: self.display_design_conditions(location) with tab2: self.plot_psychrometric_chart(location, epw_data) with tab3: self.plot_sun_shading_chart(location) with tab4: self.plot_temperature_range(location, epw_data) with tab5: self.plot_wind_rose(epw_data) else: st.info("No climate data available. Please upload an EPW file to proceed.") # Navigation buttons col1, col2 = st.columns(2) with col1: st.button("Back to Building Information", on_click=lambda: setattr(session_state, "page", "Building Information")) with col2: if self.locations: st.button("Continue to Building Components", on_click=lambda: setattr(session_state, "page", "Building Components")) else: st.button("Continue to Building Components", disabled=True) def display_design_conditions(self, location: ClimateLocation): """Display design conditions for HVAC calculations using Markdown.""" st.subheader("Design Conditions") st.markdown(f""" **Location Details:** - **Country**: {location.country} - **City**: {location.city} - **State/Province**: {location.state_province} - **Latitude**: {location.latitude}° - **Longitude**: {location.longitude}° - **Elevation**: {location.elevation} m **Climate Parameters:** - **Climate Zone**: {location.climate_zone} - **Heating Degree Days (base 18°C)**: {location.heating_degree_days} HDD - **Cooling Degree Days (base 18°C)**: {location.cooling_degree_days} CDD - **Winter Design Temperature (99.6%)**: {location.winter_design_temp} °C - **Summer Design Dry-Bulb Temp (0.4%)**: {location.summer_design_temp_db} °C - **Summer Design Wet-Bulb Temp (0.4%)**: {location.summer_design_temp_wb} °C - **Summer Daily Temperature Range**: {location.summer_daily_range} °C - **Mean Wind Speed**: {location.wind_speed} m/s - **Mean Atmospheric Pressure**: {location.pressure} Pa """) @staticmethod def assign_climate_zone(hdd: float, cdd: float, avg_humidity: float) -> str: """Assign ASHRAE 169 climate zone based on HDD, CDD, and humidity.""" if cdd > 10000: return "0A" if avg_humidity > 60 else "0B" elif cdd > 5000: return "1A" if avg_humidity > 60 else "1B" elif cdd > 2500: return "2A" if avg_humidity > 60 else "2B" elif hdd < 2000 and cdd > 1000: return "3A" if avg_humidity > 60 else "3B" if avg_humidity < 40 else "3C" elif hdd < 3000: return "4A" if avg_humidity > 60 else "4B" if avg_humidity < 40 else "4C" elif hdd < 4000: return "5A" if avg_humidity > 60 else "5B" if avg_humidity < 40 else "5C" elif hdd < 5000: return "6A" if avg_humidity > 60 else "6B" elif hdd < 7000: return "7" else: return "8" def plot_psychrometric_chart(self, location: ClimateLocation, epw_data: pd.DataFrame): """Plot psychrometric chart with ASHRAE 55 comfort zone and psychrometric lines.""" st.subheader("Psychrometric Chart") dry_bulb = pd.to_numeric(epw_data[6], errors='coerce').values humidity = pd.to_numeric(epw_data[8], errors='coerce').values valid_mask = ~np.isnan(dry_bulb) & ~np.isnan(humidity) dry_bulb = dry_bulb[valid_mask] humidity = humidity[valid_mask] # Calculate humidity ratio (kg/kg dry air) pressure = location.pressure / 1000 # kPa saturation_pressure = 6.1078 * 10 ** (7.5 * dry_bulb / (dry_bulb + 237.3)) vapor_pressure = humidity / 100 * saturation_pressure humidity_ratio = 0.62198 * vapor_pressure / (pressure - vapor_pressure) * 1000 # Convert to g/kg fig = go.Figure() # Hourly data points fig.add_trace(go.Scatter( x=dry_bulb, y=humidity_ratio, mode='markers', marker=dict(size=5, opacity=0.5, color='blue'), name='Hourly Conditions' )) # ASHRAE 55 comfort zone (simplified: 20-26°C, adjusted for humidity ratio) comfort_db = [20, 26, 26, 20, 20] comfort_rh = [30, 30, 60, 60, 30] comfort_vp = np.array(comfort_rh) / 100 * 6.1078 * 10 ** (7.5 * np.array(comfort_db) / (np.array(comfort_db) + 237.3)) comfort_hr = 0.62198 * comfort_vp / (pressure - comfort_vp) * 1000 fig.add_trace(go.Scatter( x=comfort_db, y=comfort_hr, mode='lines', line=dict(color='green', width=2), fill='toself', fillcolor='rgba(0, 255, 0, 0.2)', name='ASHRAE 55 Comfort Zone' )) # Constant humidity ratio lines (inspired by Climate Consultant) for hr in [5, 10, 15]: # g/kg db_range = np.linspace(0, 40, 100) vp = (hr / 1000 * pressure) / (0.62198 + hr / 1000) rh = vp / (6.1078 * 10 ** (7.5 * db_range / (db_range + 237.3))) * 100 hr_line = np.full_like(db_range, hr) fig.add_trace(go.Scatter( x=db_range, y=hr_line, mode='lines', line=dict(color='gray', width=1, dash='dash'), name=f'{hr} g/kg', showlegend=True )) # Constant wet-bulb temperature lines wet_bulb_temps = [10, 15, 20] for wbt in wet_bulb_temps: db_range = np.linspace(0, 40, 100) rh_range = np.linspace(5, 95, 100) wb_values = self.calculate_wet_bulb(db_range, rh_range) vp = rh_range / 100 * (6.1078 * 10 ** (7.5 * db_range / (db_range + 237.3))) hr_values = 0.62198 * vp / (pressure - vp) * 1000 mask = (wb_values >= wbt - 0.5) & (wb_values <= wbt + 0.5) if np.any(mask): fig.add_trace(go.Scatter( x=db_range[mask], y=hr_values[mask], mode='lines', line=dict(color='purple', width=1, dash='dot'), name=f'Wet-Bulb {wbt}°C', showlegend=True )) fig.update_layout( title="Psychrometric Chart", xaxis_title="Dry-Bulb Temperature (°C)", yaxis_title="Humidity Ratio (g/kg dry air)", xaxis=dict(range=[-5, 40]), # Adjusted for Geelong (3.1°C to 33.0°C) yaxis=dict(range=[0, 25]), # Adjusted for typical humidity ratios showlegend=True, template='plotly_white' ) st.plotly_chart(fig, use_container_width=True) def plot_sun_shading_chart(self, location: ClimateLocation): """Plot sun path chart for summer and winter solstices, inspired by Climate Consultant.""" st.subheader("Sun Shading Chart") dates = [ datetime(2025, 6, 21), # Winter solstice (Southern Hemisphere) datetime(2025, 12, 21) # Summer solstice (Southern Hemisphere) ] times = pd.date_range(start="2025-01-01 00:00", end="2025-01-01 23:00", freq='H') solar_data = [] for date in dates: solpos = pvlib.solarposition.get_solarposition( time=[date.replace(hour=t.hour, minute=t.minute) for t in times], latitude=location.latitude, longitude=location.longitude, altitude=location.elevation ) solar_data.append({ 'date': date.strftime('%Y-%m-%d'), 'azimuth': solpos['azimuth'].values, 'altitude': solpos['elevation'].values }) fig = go.Figure() colors = ['orange', 'blue'] # Summer = orange, Winter = blue labels = ['Summer Solstice (Dec 21)', 'Winter Solstice (Jun 21)'] for i, data in enumerate(solar_data): fig.add_trace(go.Scatterpolar( r=data['altitude'], theta=data['azimuth'], mode='lines+markers', name=labels[i], line=dict(color=colors[i], width=2), marker=dict(size=6, color=colors[i]), opacity=0.8 )) fig.update_layout( title="Sun Path Diagram", polar=dict( radialaxis=dict( range=[0, 90], tickvals=[0, 30, 60, 90], ticktext=["0°", "30°", "60°", "90°"], title="Altitude (degrees)" ), angularaxis=dict( direction="clockwise", rotation=90, tickvals=[0, 90, 180, 270], ticktext=["N", "E", "S", "W"] ) ), showlegend=True, template='plotly_white' ) st.plotly_chart(fig, use_container_width=True) def plot_temperature_range(self, location: ClimateLocation, epw_data: pd.DataFrame): """Plot monthly temperature ranges with design conditions.""" st.subheader("Monthly Temperature Range") months = pd.to_numeric(epw_data[1], errors='coerce').values dry_bulb = pd.to_numeric(epw_data[6], errors='coerce').values month_names = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] temps_min = [] temps_max = [] temps_avg = [] for i in range(1, 13): month_mask = (months == i) temps_min.append(round(np.nanmin(dry_bulb[month_mask]), 1)) temps_max.append(round(np.nanmax(dry_bulb[month_mask]), 1)) temps_avg.append(round(np.nanmean(dry_bulb[month_mask]), 1)) fig = go.Figure() fig.add_trace(go.Scatter( x=list(range(1, 13)), y=temps_max, mode='lines', name='Max Temperature', line=dict(color='red', dash='dash'), opacity=0.5 )) fig.add_trace(go.Scatter( x=list(range(1, 13)), y=temps_min, mode='lines', name='Min Temperature', line=dict(color='red', dash='dash'), opacity=0.5, fill='tonexty', fillcolor='rgba(255, 0, 0, 0.1)' )) fig.add_trace(go.Scatter( x=list(range(1, 13)), y=temps_avg, mode='lines+markers', name='Avg Temperature', line=dict(color='red'), marker=dict(size=8) )) # Add design temperatures fig.add_hline(y=location.winter_design_temp, line_dash="dot", line_color="blue", annotation_text="Winter Design Temp", annotation_position="top left") fig.add_hline(y=location.summer_design_temp_db, line_dash="dot", line_color="orange", annotation_text="Summer Design Temp (DB)", annotation_position="bottom left") fig.update_layout( title="Monthly Temperature Profile", xaxis_title="Month", yaxis_title="Temperature (°C)", xaxis=dict(tickmode='array', tickvals=list(range(1, 13)), ticktext=month_names), legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01), showlegend=True, template='plotly_white' ) st.plotly_chart(fig, use_container_width=True) def plot_wind_rose(self, epw_data: pd.DataFrame): """Plot wind rose diagram with improved clarity, inspired by Climate Consultant.""" st.subheader("Wind Rose") wind_speed = pd.to_numeric(epw_data[21], errors='coerce').values wind_direction = pd.to_numeric(epw_data[20], errors='coerce').values valid_mask = ~np.isnan(wind_speed) & ~np.isnan(wind_direction) wind_speed = wind_speed[valid_mask] wind_direction = wind_direction[valid_mask] # Bin data with 8 directions and tailored speed bins (based on Geelong’s mean wind speed of 4.0 m/s) speed_bins = [0, 2, 4, 6, 8, np.inf] direction_bins = np.linspace(0, 360, 9)[:-1] speed_labels = ['0-2 m/s', '2-4 m/s', '4-6 m/s', '6-8 m/s', '8+ m/s'] direction_labels = ['N', 'NE', 'E', 'SE', 'S', 'SW', 'W', 'NW'] hist = np.histogram2d( wind_direction, wind_speed, bins=[direction_bins, speed_bins], density=True )[0] hist = hist * 100 # Convert to percentage fig = go.Figure() colors = ['#E6F0FF', '#B3D1FF', '#80B2FF', '#4D94FF', '#1A75FF'] # Light to dark blue gradient for i, speed_label in enumerate(speed_labels): fig.add_trace(go.Barpolar( r=hist[:, i], theta=direction_bins, width=45, name=speed_label, marker=dict(color=colors[i]), opacity=0.8 )) fig.update_layout( title="Wind Rose", polar=dict( radialaxis=dict( tickvals=[0, 5, 10, 15], ticktext=["0%", "5%", "10%", "15%"], title="Frequency (%)" ), angularaxis=dict( direction="clockwise", rotation=90, tickvals=direction_bins, ticktext=direction_labels ) ), showlegend=True, template='plotly_white' ) st.plotly_chart(fig, use_container_width=True) def export_to_json(self, file_path: str) -> None: """Export all climate data to a JSON file.""" data = {loc_id: loc.to_dict() for loc_id, loc in self.locations.items()} with open(file_path, 'w') as f: json.dump(data, f, indent=4) @classmethod def from_json(cls, file_path: str) -> 'ClimateData': """Load climate data from a JSON file.""" with open(file_path, 'r') as f: data = json.load(f) climate_data = cls() for loc_id, loc_dict in data.items(): location = ClimateLocation(epw_file=pd.DataFrame(), **loc_dict) climate_data.add_location(location) return climate_data if __name__ == "__main__": climate_data = ClimateData() session_state = {"building_info": {"country": "Australia", "city": "Geelong"}, "page": "Climate Data"} climate_data.display_climate_input(session_state)