File size: 6,510 Bytes
b64d8b8 917701a b64d8b8 917701a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
import streamlit as st
from PIL import Image
import os
from dotenv import load_dotenv
import requests
import base64
from io import BytesIO
import json
import re
# Load environment variables
load_dotenv()
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
# Placeholder for AI image analysis (fallback)
def analyze_image_placeholder():
return {
"area_m2": 100,
"orientation_deg": 180,
"shading_percent": 10,
"obstructions": ["chimney"]
}
# OpenRouter API image analysis
def analyze_image(image=None):
if image is None:
return analyze_image_placeholder()
try:
# Convert PIL image to base64
buffered = BytesIO()
image.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode()
# Call OpenRouter API
url = "https://openrouter.ai/api/v1/chat/completions"
headers = {
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "opengvlab/internvl3-14b:free",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"Analyze this rooftop image for solar potential. "
"Respond ONLY with valid JSON, no explanation, with these keys: "
"area_m2, orientation_deg, shading_percent, obstructions."
)
},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{img_str}"}
}
]
}
]
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
st.write("API raw response:", content) # For debugging
try:
# Extract JSON block (from first { to last })
match = re.search(r"\{.*\}", content, re.DOTALL)
if not match:
raise json.JSONDecodeError("No JSON object found", content, 0)
json_str = match.group(0)
# Remove comments (// ...)
json_str = re.sub(r"//.*", "", json_str)
# Parse JSON
parsed = json.loads(json_str)
# Map fields to expected format
if "solar_analysis" in parsed:
sa = parsed["solar_analysis"]
result_dict = {
"area_m2": sa.get("total_area", {}).get("suitable_rooftops_area", 100),
"orientation_deg": sa.get("orientation", 180),
"shading_percent": sa.get("shading_percentage", 10),
"obstructions": sa.get("obstructions", []),
}
else:
result_dict = parsed # fallback, in case structure matches directly
# Validate required fields
required = ["area_m2", "orientation_deg", "shading_percent", "obstructions"]
if all(key in result_dict for key in required):
return result_dict
else:
st.warning("API response missing required fields. Using placeholder data.")
return analyze_image_placeholder()
except Exception as e:
st.warning(f"Invalid JSON from API. Using placeholder data. ({e})")
return analyze_image_placeholder()
else:
st.error(f"API error: {response.text}")
return analyze_image_placeholder()
except Exception as e:
st.error(f"Error processing image: {str(e)}")
return analyze_image_placeholder()
def orientation_factor(orientation):
if 165 <= orientation <= 195: # South-facing
return 1.0
elif 75 <= orientation <= 105 or 255 <= orientation <= 285: # East/West
return 0.8
else: # Other orientations
return 0.6
# Calculate solar potential (kWh/year)
def calculate_solar_potential(area, orientation, shading, insolation=5):
efficiency = 0.2 # Panel efficiency
usable_area = area * (1 - shading / 100)
orientation_adj = usable_area * orientation_factor(orientation)
annual_kwh = orientation_adj * insolation * 365 * efficiency
return round(annual_kwh, 2)
# Calculate ROI
def calculate_roi(kwh, cost_per_watt=3, incentive=0.3, electricity_rate=0.12):
system_size_w = kwh / (5 * 365) * 1000 # Convert kWh to system size
total_cost = system_size_w * cost_per_watt
cost_after_incentive = total_cost * (1 - incentive)
payback_years = cost_after_incentive / (kwh * electricity_rate)
return round(cost_after_incentive, 2), round(payback_years, 1)
# Streamlit app
st.title("Solar Rooftop Analysis Tool")
# Image upload
uploaded_file = st.file_uploader("Upload satellite image of rooftop", type=["png", "jpg", "jpeg"])
if uploaded_file is not None:
image = Image.open(uploaded_file)
st.image(image, caption="Uploaded Rooftop Image", use_column_width=True)
result = analyze_image(image)
else:
st.write("No image uploaded. Using placeholder data: 100m², south-facing, 10% shading.")
result = analyze_image()
# Display analysis results
st.write("### Rooftop Analysis")
st.json(result)
# Calculate and display solar potential
kwh = calculate_solar_potential(
result["area_m2"], result["orientation_deg"], result["shading_percent"]
)
st.write(f"**Estimated Annual Energy Production**: {kwh} kWh")
# Calculate and display ROI
cost, payback = calculate_roi(kwh)
st.write(f"**Estimated Cost (after 30% incentive)**: ${cost}")
st.write(f"**Payback Period**: {payback} years")
# Installation recommendations
st.write("### Installation Recommendations")
st.write("- **Panel Type**: Monocrystalline (20% efficiency)")
st.write(f"- **Number of Panels**: ~{int(result['area_m2'] / 2)} (2m² per panel)")
st.write("- **Mounting**: Flush mount, south-facing")
st.write("- **Maintenance**: Annual cleaning, monitor via app")
st.write("- **Compliance**: Follow NEC 2020 standards, check local net metering policies") |