mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
tidying up solar api code
This commit is contained in:
Khalim Conn-Kowlessar
parent
e4aa4cbb2f
commit
46f513a4f2
4 changed files with 249 additions and 183 deletions
|
|
@ -631,17 +631,23 @@ class Property:
|
|||
"""
|
||||
self.solar_panel_configuration = solar_panel_configuration
|
||||
|
||||
if not self.roof["is_flat"]:
|
||||
default_roof_area = estimate_pitched_roof_area(
|
||||
floor_area=self.insulation_floor_area,
|
||||
floor_height=self.floor_height
|
||||
)
|
||||
else:
|
||||
default_roof_area = self.insulation_floor_area
|
||||
|
||||
# We also set the roof area
|
||||
if roof_area is None:
|
||||
if self.roof["is_flat"]:
|
||||
self.roof_area = estimate_pitched_roof_area(
|
||||
floor_area=self.insulation_floor_area,
|
||||
floor_height=self.floor_height
|
||||
)
|
||||
else:
|
||||
self.roof_area = self.insulation_floor_area
|
||||
|
||||
self.roof_area = default_roof_area
|
||||
else:
|
||||
# Perform a comparison between the default_roof_area and roof_area
|
||||
difference = abs(default_roof_area - roof_area)
|
||||
if difference / default_roof_area > 0.1:
|
||||
raise Exception("Investigate difference in roof area")
|
||||
|
||||
self.roof_area = roof_area
|
||||
|
||||
def set_current_energy_bill(self, kwh_client, kwh_predictions):
|
||||
|
|
|
|||
|
|
@ -1,16 +1,22 @@
|
|||
import time
|
||||
import requests
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from recommendations.Costs import MCS_SOLAR_PV_COST_DATA
|
||||
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
|
||||
import requests
|
||||
from typing import List
|
||||
from functools import lru_cache
|
||||
import time
|
||||
from backend.app.db.functions.solar_functions import get_solar_data, store_batch_data
|
||||
from utils.logger import setup_logger
|
||||
from sklearn.preprocessing import MinMaxScaler
|
||||
from recommendations.Costs import Costs
|
||||
from tqdm import tqdm
|
||||
from math import sin, cos, sqrt, atan2, radians
|
||||
|
||||
from utils.logger import setup_logger
|
||||
from recommendations.Costs import Costs, MCS_SOLAR_PV_COST_DATA
|
||||
from etl.bill_savings.EnergyConsumptionModel import EnergyConsumptionModel
|
||||
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
|
||||
from backend.Property import Property
|
||||
from backend.app.db.functions.solar_functions import get_solar_data, store_batch_data
|
||||
import backend.app.assumptions as assumptions
|
||||
from backend.app.plan.schemas import PlanTriggerRequest
|
||||
|
||||
logger = setup_logger()
|
||||
|
||||
|
||||
|
|
@ -590,6 +596,215 @@ class GoogleSolarApi:
|
|||
# implies we should do this
|
||||
self.double_property = True
|
||||
|
||||
@staticmethod
|
||||
def prepare_input_data(
|
||||
input_properties: List[Property],
|
||||
energy_consumption_client: EnergyConsumptionModel,
|
||||
body: PlanTriggerRequest
|
||||
):
|
||||
"""
|
||||
:param input_properties: List of properties
|
||||
:param energy_consumption_client: EnergyConsumptionModel instance
|
||||
:param body: PlanTriggerRequest instance
|
||||
This sets up the data required to make the solar api request
|
||||
:return:
|
||||
"""
|
||||
|
||||
building_solar_config = [
|
||||
{
|
||||
"building_id": p.building_id,
|
||||
"longitude": p.spatial["longitude"],
|
||||
"latitude": p.spatial["latitude"],
|
||||
# Energy consumption is adjusted for the property's expected post retrofit state
|
||||
# We set the target rating to EPC C, which is the typical EPC rating we would expect the
|
||||
# property to achieve post retrofit of just the fabric
|
||||
"energy_consumption": energy_consumption_client.estimate_new_consumption(
|
||||
current_energy_efficiency=p.data["current-energy-efficiency"],
|
||||
target_efficiency="69",
|
||||
current_consumption=p.estimate_electrical_consumption(
|
||||
assumed_ashp_efficiency=assumptions.AVERAGE_ASHP_EFFICIENCY, exclusions=body.exclusions
|
||||
)
|
||||
),
|
||||
"property_id": p.id,
|
||||
"uprn": p.uprn
|
||||
} for p in input_properties if p.building_id is not None
|
||||
]
|
||||
unit_solar_config = [
|
||||
{
|
||||
"longitude": p.spatial["longitude"],
|
||||
"latitude": p.spatial["latitude"],
|
||||
# Energy consumption is adjusted for the property's expected post retrofit state
|
||||
# We set the target rating to EPC C, which is the typical EPC rating we would expect the
|
||||
# property to achieve post retrofit of just the fabric
|
||||
"energy_consumption": energy_consumption_client.estimate_new_consumption(
|
||||
current_energy_efficiency=p.data["current-energy-efficiency"],
|
||||
target_efficiency="69",
|
||||
current_consumption=p.estimate_electrical_consumption(
|
||||
assumed_ashp_efficiency=assumptions.AVERAGE_ASHP_EFFICIENCY, exclusions=body.exclusions
|
||||
),
|
||||
),
|
||||
"property_id": p.id,
|
||||
"uprn": p.uprn
|
||||
} for p in input_properties if p.building_id is None
|
||||
]
|
||||
|
||||
return building_solar_config, unit_solar_config
|
||||
|
||||
@classmethod
|
||||
def building_solar_analysis(
|
||||
cls, building_solar_config: List, input_properties: List[Property], session, google_solar_api_key: str
|
||||
):
|
||||
"""
|
||||
Perform the solar analysis for the building level
|
||||
:param building_solar_config: List of building solar configurations
|
||||
:param input_properties: List of properties
|
||||
:param session: Database session
|
||||
:param google_solar_api_key: Google Solar API key
|
||||
:return:
|
||||
"""
|
||||
|
||||
if not building_solar_config:
|
||||
return input_properties
|
||||
|
||||
# Find the unique longitude and latitude pairs for each building id
|
||||
unique_coordinates = {}
|
||||
building_uprns = {}
|
||||
for entry in building_solar_config:
|
||||
building_id = entry['building_id']
|
||||
coordinate_pair = {'longitude': entry['longitude'], 'latitude': entry['latitude']}
|
||||
|
||||
if building_id not in unique_coordinates:
|
||||
unique_coordinates[building_id] = []
|
||||
|
||||
if coordinate_pair not in unique_coordinates[building_id]:
|
||||
unique_coordinates[building_id].append(coordinate_pair)
|
||||
|
||||
if building_id not in building_uprns:
|
||||
building_uprns[building_id] = []
|
||||
|
||||
if entry['uprn'] not in building_uprns[building_id]:
|
||||
building_uprns[building_id].append(
|
||||
{
|
||||
"uprn": entry['uprn'], "longitude": entry['longitude'], "latitude": entry['latitude']
|
||||
}
|
||||
)
|
||||
|
||||
solar_panel_configuration = {}
|
||||
for building_id, coordinates in unique_coordinates.items():
|
||||
if len(coordinates) > 1:
|
||||
raise NotImplementedError("more than one coordinate for a building - handle me")
|
||||
|
||||
coordinates = coordinates[0]
|
||||
energy_consumption = sum(
|
||||
[entry['energy_consumption'] for entry in building_solar_config if entry['building_id'] == building_id]
|
||||
)
|
||||
solar_api_client = cls(api_key=google_solar_api_key)
|
||||
solar_api_client.get(
|
||||
longitude=coordinates["longitude"],
|
||||
latitude=coordinates["latitude"],
|
||||
energy_consumption=energy_consumption,
|
||||
is_building=True,
|
||||
session=session
|
||||
)
|
||||
solar_panel_configuration[building_id] = {
|
||||
"insights_data": solar_api_client.insights_data,
|
||||
"panel_performance": solar_api_client.panel_performance,
|
||||
"n_units": len([entry for entry in building_solar_config if entry['building_id'] == building_id])
|
||||
}
|
||||
|
||||
# Store the data in the database
|
||||
# TODO: Rather than just doing a straight insert, we should overwrite what's already there if it
|
||||
# exists
|
||||
solar_api_client.save_to_db(
|
||||
session=session, uprns_to_location=building_uprns[building_id], scenario_type="building"
|
||||
)
|
||||
|
||||
# Insert this into the properties that have this building id
|
||||
for p in input_properties:
|
||||
if p.building_id == building_id:
|
||||
unit_solar_panel_configuration = solar_panel_configuration[building_id].copy()
|
||||
|
||||
unit_solar_panel_configuration["unit_share_of_energy"] = (
|
||||
[x for x in building_solar_config if x["property_id"] == p.id][0]["energy_consumption"] /
|
||||
energy_consumption
|
||||
)
|
||||
p.set_solar_panel_configuration(unit_solar_panel_configuration)
|
||||
|
||||
return input_properties
|
||||
|
||||
@classmethod
|
||||
def unit_solar_analysis(
|
||||
cls, unit_solar_config: List, input_properties: List[Property], session, body, google_solar_api_key: str
|
||||
):
|
||||
|
||||
if not unit_solar_config:
|
||||
return input_properties
|
||||
|
||||
# Model the solar potential at the property level
|
||||
for unit in tqdm(unit_solar_config):
|
||||
|
||||
# We don't need to do this if we have global inclusions that don't include solar
|
||||
if body.inclusions:
|
||||
if "solar_pv" not in body.inclusions:
|
||||
continue
|
||||
|
||||
property_instance = [p for p in input_properties if p.id == unit["property_id"]][0]
|
||||
# At this level, we check if the property is suitable for solar and if now, skip
|
||||
# Or if we have a solar non-invasive recommendation
|
||||
if (
|
||||
(not property_instance.is_solar_pv_valid()) or
|
||||
[r for r in property_instance.non_invasive_recommendations if r["type"] == "solar_pv"]
|
||||
):
|
||||
continue
|
||||
|
||||
if unit["longitude"] is None or unit["latitude"] is None:
|
||||
# At this point, we've checked that solar PV is valid, and so we provide some defaults
|
||||
|
||||
property_instance.set_solar_panel_configuration(
|
||||
solar_panel_configuration={
|
||||
"insights_data": None,
|
||||
"panel_performance": cls.default_panel_performance(property_instance=property_instance),
|
||||
"unit_share_of_energy": 1
|
||||
},
|
||||
roof_area=None
|
||||
)
|
||||
continue
|
||||
|
||||
solar_api_client = cls(api_key=google_solar_api_key)
|
||||
solar_api_client.get(
|
||||
longitude=unit["longitude"],
|
||||
latitude=unit["latitude"],
|
||||
energy_consumption=unit["energy_consumption"],
|
||||
is_building=False,
|
||||
session=session,
|
||||
uprn=unit["uprn"],
|
||||
property_instance=property_instance
|
||||
)
|
||||
|
||||
# Store the data in the database
|
||||
solar_api_client.save_to_db(
|
||||
session=session,
|
||||
uprns_to_location=[
|
||||
{
|
||||
"uprn": property_instance.uprn,
|
||||
"longitude": property_instance.spatial["longitude"],
|
||||
"latitude": property_instance.spatial["latitude"]
|
||||
}
|
||||
],
|
||||
scenario_type="unit"
|
||||
)
|
||||
|
||||
property_instance.set_solar_panel_configuration(
|
||||
solar_panel_configuration={
|
||||
"insights_data": solar_api_client.insights_data,
|
||||
"panel_performance": solar_api_client.panel_performance,
|
||||
"unit_share_of_energy": 1
|
||||
},
|
||||
roof_area=solar_api_client.roof_area
|
||||
)
|
||||
|
||||
return input_properties
|
||||
|
||||
@classmethod
|
||||
def default_panel_performance(cls, property_instance):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -513,179 +513,24 @@ async def trigger_plan(body: PlanTriggerRequest):
|
|||
[p.set_features(cleaned=cleaned, kwh_client=kwh_client, kwh_predictions=kwh_preds) for p in input_properties]
|
||||
logger.info("Performing solar analysis")
|
||||
|
||||
# TODO: Tidy this up
|
||||
# TODO: If a property is semi-detached, we might get roof surfaces for the main building + the neighbour
|
||||
# TODO: If we can't get high image quality, should we use the solar API? Maybe just for semi-detached units with
|
||||
# extensions, since it doesn't seem to do a great job
|
||||
# TODO: For simple properties, we should do a comparison/check between the solar API's roof area and the
|
||||
# basic estimate of roof area
|
||||
|
||||
building_ids = [
|
||||
{
|
||||
"building_id": p.building_id,
|
||||
"longitude": p.spatial["longitude"],
|
||||
"latitude": p.spatial["latitude"],
|
||||
# Energy consumption is adjusted for the property's expected post retrofit state
|
||||
# We set the target rating to EPC C, which is the typical EPC rating we would expect the
|
||||
# property to achieve post retrofit of just the fabric
|
||||
"energy_consumption": energy_consumption_client.estimate_new_consumption(
|
||||
current_energy_efficiency=p.data["current-energy-efficiency"],
|
||||
target_efficiency="69",
|
||||
current_consumption=p.estimate_electrical_consumption(
|
||||
assumed_ashp_efficiency=assumptions.AVERAGE_ASHP_EFFICIENCY, exclusions=body.exclusions
|
||||
)
|
||||
),
|
||||
"property_id": p.id,
|
||||
"uprn": p.uprn
|
||||
} for p in input_properties if p.building_id is not None
|
||||
]
|
||||
individual_units = [
|
||||
{
|
||||
"longitude": p.spatial["longitude"],
|
||||
"latitude": p.spatial["latitude"],
|
||||
# Energy consumption is adjusted for the property's expected post retrofit state
|
||||
# We set the target rating to EPC C, which is the typical EPC rating we would expect the
|
||||
# property to achieve post retrofit of just the fabric
|
||||
"energy_consumption": energy_consumption_client.estimate_new_consumption(
|
||||
current_energy_efficiency=p.data["current-energy-efficiency"],
|
||||
target_efficiency="69",
|
||||
current_consumption=p.estimate_electrical_consumption(
|
||||
assumed_ashp_efficiency=assumptions.AVERAGE_ASHP_EFFICIENCY, exclusions=body.exclusions
|
||||
),
|
||||
),
|
||||
"property_id": p.id,
|
||||
"uprn": p.uprn
|
||||
} for p in input_properties if p.building_id is None
|
||||
]
|
||||
if building_ids:
|
||||
# Find the unique longitude and latitude pairs for each building id
|
||||
unique_coordinates = {}
|
||||
building_uprns = {}
|
||||
for entry in building_ids:
|
||||
building_id = entry['building_id']
|
||||
coordinate_pair = {'longitude': entry['longitude'], 'latitude': entry['latitude']}
|
||||
building_solar_config, unit_solar_config = GoogleSolarApi.prepare_input_data(
|
||||
input_properties=input_properties,
|
||||
energy_consumption_client=energy_consumption_client,
|
||||
body=body
|
||||
)
|
||||
|
||||
if building_id not in unique_coordinates:
|
||||
unique_coordinates[building_id] = []
|
||||
|
||||
if coordinate_pair not in unique_coordinates[building_id]:
|
||||
unique_coordinates[building_id].append(coordinate_pair)
|
||||
|
||||
if building_id not in building_uprns:
|
||||
building_uprns[building_id] = []
|
||||
|
||||
if entry['uprn'] not in building_uprns[building_id]:
|
||||
building_uprns[building_id].append(
|
||||
{
|
||||
"uprn": entry['uprn'], "longitude": entry['longitude'], "latitude": entry['latitude']
|
||||
}
|
||||
)
|
||||
|
||||
solar_panel_configuration = {}
|
||||
for building_id, coordinates in unique_coordinates.items():
|
||||
if len(coordinates) > 1:
|
||||
raise NotImplementedError("more than one coordinate for a building - handle me")
|
||||
|
||||
coordinates = coordinates[0]
|
||||
energy_consumption = sum(
|
||||
[entry['energy_consumption'] for entry in building_ids if entry['building_id'] == building_id]
|
||||
)
|
||||
solar_api_client = GoogleSolarApi(api_key=get_settings().GOOGLE_SOLAR_API_KEY)
|
||||
solar_api_client.get(
|
||||
longitude=coordinates["longitude"],
|
||||
latitude=coordinates["latitude"],
|
||||
energy_consumption=energy_consumption,
|
||||
is_building=True,
|
||||
session=session
|
||||
)
|
||||
solar_panel_configuration[building_id] = {
|
||||
"insights_data": solar_api_client.insights_data,
|
||||
"panel_performance": solar_api_client.panel_performance,
|
||||
"n_units": len([entry for entry in building_ids if entry['building_id'] == building_id])
|
||||
}
|
||||
|
||||
# Store the data in the database
|
||||
# TODO: Rather than just doing a straight insert, we should overwrite what's already there if it
|
||||
# exists
|
||||
solar_api_client.save_to_db(
|
||||
session=session, uprns_to_location=building_uprns[building_id], scenario_type="building"
|
||||
)
|
||||
|
||||
# Insert this into the properties that have this building id
|
||||
for p in input_properties:
|
||||
if p.building_id == building_id:
|
||||
unit_solar_panel_configuration = solar_panel_configuration[building_id].copy()
|
||||
|
||||
unit_solar_panel_configuration["unit_share_of_energy"] = (
|
||||
[x for x in building_ids if x["property_id"] == p.id][0]["energy_consumption"] /
|
||||
energy_consumption
|
||||
)
|
||||
p.set_solar_panel_configuration(unit_solar_panel_configuration)
|
||||
if individual_units:
|
||||
# Model the solar potential at the property level
|
||||
for unit in tqdm(individual_units):
|
||||
|
||||
# TODO: Tidy up this code
|
||||
# We don't need to do this if we have global inclusions that don't include solar
|
||||
if body.inclusions:
|
||||
if "solar_pv" not in body.inclusions:
|
||||
continue
|
||||
|
||||
property_instance = [p for p in input_properties if p.id == unit["property_id"]][0]
|
||||
# At this level, we check if the property is suitable for solar and if now, skip
|
||||
if not property_instance.is_solar_pv_valid():
|
||||
continue
|
||||
|
||||
if unit["longitude"] is None or unit["latitude"] is None:
|
||||
# At this point, we've checked that solar PV is valid, and so we provide some defaults
|
||||
|
||||
property_instance.set_solar_panel_configuration(
|
||||
solar_panel_configuration={
|
||||
"insights_data": None,
|
||||
"panel_performance": GoogleSolarApi.default_panel_performance(
|
||||
property_instance=property_instance
|
||||
),
|
||||
"unit_share_of_energy": 1
|
||||
},
|
||||
roof_area=None
|
||||
)
|
||||
continue
|
||||
|
||||
# We check if we have a solar non-invasive recommendation
|
||||
if [r for r in property_instance.non_invasive_recommendations if r["type"] == "solar_pv"]:
|
||||
continue
|
||||
solar_api_client = GoogleSolarApi(api_key=get_settings().GOOGLE_SOLAR_API_KEY)
|
||||
solar_api_client.get(
|
||||
longitude=unit["longitude"],
|
||||
latitude=unit["latitude"],
|
||||
energy_consumption=unit["energy_consumption"],
|
||||
is_building=False,
|
||||
session=session,
|
||||
uprn=unit["uprn"],
|
||||
property_instance=property_instance
|
||||
)
|
||||
|
||||
# Store the data in the database
|
||||
solar_api_client.save_to_db(
|
||||
session=session,
|
||||
uprns_to_location=[
|
||||
{
|
||||
"uprn": property_instance.uprn,
|
||||
"longitude": property_instance.spatial["longitude"],
|
||||
"latitude": property_instance.spatial["latitude"]
|
||||
}
|
||||
],
|
||||
scenario_type="unit"
|
||||
)
|
||||
|
||||
property_instance.set_solar_panel_configuration(
|
||||
solar_panel_configuration={
|
||||
"insights_data": solar_api_client.insights_data,
|
||||
"panel_performance": solar_api_client.panel_performance,
|
||||
"unit_share_of_energy": 1
|
||||
},
|
||||
roof_area=solar_api_client.roof_area
|
||||
)
|
||||
input_properties = GoogleSolarApi.building_solar_analysis(
|
||||
building_solar_config=building_solar_config,
|
||||
input_properties=input_properties,
|
||||
session=session,
|
||||
google_solar_api_key=get_settings().GOOGLE_SOLAR_API_KEY
|
||||
)
|
||||
|
||||
logger.info("Identifying property recommendations")
|
||||
recommendations = {}
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import numpy as np
|
|||
import pandas as pd
|
||||
|
||||
from recommendations.Costs import Costs
|
||||
from recommendations.recommendation_utils import override_costs, esimtate_pitched_roof_area
|
||||
from recommendations.recommendation_utils import override_costs, estimate_pitched_roof_area
|
||||
|
||||
|
||||
class SolarPvRecommendations:
|
||||
|
|
@ -174,7 +174,7 @@ class SolarPvRecommendations:
|
|||
if self.property.roof["is_flat"]:
|
||||
roof_area = self.property.insulation_floor_area
|
||||
else:
|
||||
roof_area = esimtate_pitched_roof_area(
|
||||
roof_area = estimate_pitched_roof_area(
|
||||
floor_area=self.property.insulation_floor_area, floor_height=self.property.data["floor-height"]
|
||||
)
|
||||
solar_configurations = pd.DataFrame(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue