mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
96 lines
4 KiB
Python
96 lines
4 KiB
Python
import datetime
|
|
import pytz
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from backend.app.db.models.solar import Solar, SolarScenario
|
|
|
|
|
|
def get_solar_data(session: Session, longitude: float = None, latitude: float = None, uprn: str = None):
|
|
"""
|
|
This function will fetch data from the solar table based on longitude and latitude or UPRN.
|
|
:param session: The database session
|
|
:param longitude: The longitude to search for
|
|
:param latitude: The latitude to search for
|
|
:param uprn: The UPRN to search for (overrides longitude and latitude if provided)
|
|
:return: The google_api_response and updated_at fields
|
|
"""
|
|
try:
|
|
if uprn:
|
|
# Search by UPRN
|
|
solar_data = session.query(Solar.google_api_response, Solar.updated_at).filter_by(uprn=uprn).one()
|
|
else:
|
|
# Search by longitude and latitude
|
|
solar_data = session.query(Solar.google_api_response, Solar.updated_at).filter(
|
|
Solar.longitude == longitude,
|
|
Solar.latitude == latitude
|
|
).one()
|
|
|
|
# Check if updated_at is more than 6 months old
|
|
six_months_ago = datetime.datetime.now(pytz.utc) - datetime.timedelta(days=6 * 30) # Approximate 6 months
|
|
is_outdated = solar_data.updated_at < six_months_ago
|
|
|
|
return solar_data.google_api_response, solar_data.updated_at, is_outdated
|
|
|
|
except NoResultFound:
|
|
return None, None, False
|
|
|
|
|
|
def store_batch_data(session: Session, api_data: dict, uprns_to_location: list, scenarios_data: list):
|
|
"""
|
|
This function will store the API data to the solar table against all of the UPRNs with longitude and latitude.
|
|
:param session: The database session
|
|
:param api_data: The API data to store
|
|
:param uprns_to_location: A list of dictionaries containing uprn, longitude, and latitude
|
|
:param scenarios_data: A list of dictionaries containing scenario data for each UPRN
|
|
"""
|
|
try:
|
|
|
|
# Insert data into the Solar table and get the IDs
|
|
solar_records = []
|
|
for data in uprns_to_location:
|
|
solar_record = Solar(
|
|
uprn=data['uprn'],
|
|
longitude=data['longitude'],
|
|
latitude=data['latitude'],
|
|
google_api_response=api_data,
|
|
updated_at=datetime.datetime.now(pytz.utc)
|
|
)
|
|
solar_records.append(solar_record)
|
|
session.add(solar_record)
|
|
|
|
session.flush() # Flush to get the IDs generated
|
|
|
|
for record in solar_records:
|
|
session.refresh(record) # Refresh to populate the ID fields
|
|
|
|
# Retrieve the IDs of the inserted records
|
|
inserted_ids = {record.uprn: record.id for record in solar_records}
|
|
|
|
# Prepare the data for SolarScenario
|
|
scenario_records = []
|
|
for data in uprns_to_location:
|
|
solar_id = inserted_ids.get(data['uprn'])
|
|
for scenario in scenarios_data:
|
|
scenario_record = SolarScenario(
|
|
solar_id=solar_id,
|
|
scenario_type=scenario['scenario_type'],
|
|
number_panels=scenario['number_panels'],
|
|
array_kwhp=scenario['array_kwhp'],
|
|
lifetime_dc_kwh=scenario['lifetime_dc_kwh'],
|
|
yearly_dc_kwh=scenario['yearly_dc_kwh'],
|
|
lifetime_ac_kwh=scenario.get('lifetime_ac_kwh'), # Optional field
|
|
yearly_ac_kwh=scenario.get('yearly_ac_kwh'), # Optional field
|
|
cost=scenario['cost'],
|
|
expected_payback_years=scenario.get('expected_payback_years'), # Optional field
|
|
panelled_roof_area=scenario['panelled_roof_area'],
|
|
is_default=scenario['is_default']
|
|
)
|
|
scenario_records.append(scenario_record)
|
|
|
|
# Insert data into the SolarScenario table
|
|
session.bulk_save_objects(scenario_records)
|
|
session.commit()
|
|
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise e
|