mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
import re
|
|
from datetime import datetime
|
|
from tqdm import tqdm
|
|
|
|
import pandas as pd
|
|
|
|
from utils.s3 import list_files_in_s3_folder, read_pickle_from_s3, save_dataframe_to_s3_parquet
|
|
|
|
# These columns we co-erce to strings before saving
|
|
PROBLEMATIC_COLUMNS = ["main-heating-controls", "floor-level"]
|
|
|
|
|
|
def extract_kwh_value(text):
|
|
"""
|
|
Extract the numerical kWh value from a given string.
|
|
|
|
:param text: The input string containing the kWh value.
|
|
:return: The extracted numerical kWh value as an integer.
|
|
"""
|
|
# Use regular expression to find the numerical value followed by "kWh per year"
|
|
match = re.search(r'([\d,]+) kWh per year', text)
|
|
|
|
if match:
|
|
# Remove commas from the extracted value and convert to integer
|
|
kwh_value = int(match.group(1).replace(',', ''))
|
|
return kwh_value
|
|
else:
|
|
# If no match is found, return None or raise an exception
|
|
return None
|
|
|
|
|
|
def app():
|
|
"""
|
|
Given the files written in our datalake in s3, this application will collate the data into a single file
|
|
and store it back in s3 for analysis
|
|
:return:
|
|
"""
|
|
|
|
# Firstly, list all of the saved files in s3
|
|
data_files = list_files_in_s3_folder(bucket_name="retrofit-datalake-dev", folder_name="energy_consumption_data")
|
|
|
|
run_date = datetime.now().strftime("%Y-%m-%d")
|
|
|
|
complete_data = []
|
|
for files in tqdm(data_files):
|
|
dataset_run_date = files.split("/")[-1].split(".")[0]
|
|
# Extract the date from the file name
|
|
dataset_run_date = pd.Timestamp(dataset_run_date)
|
|
|
|
# Load the data from the file
|
|
data = read_pickle_from_s3(bucket_name="retrofit-datalake-dev", s3_file_name=files)
|
|
|
|
# We check that the retrieved energy consumption sufficiently matches the EPC data
|
|
internal_dataset = []
|
|
for x in data:
|
|
epc_data = x["epc"]
|
|
epc_sap = epc_data["current-energy-efficiency"]
|
|
epc_potential_sap = epc_data["potential-energy-efficiency"]
|
|
# Make sure this matches the extracted sap
|
|
if int(epc_sap) != int(x["current_epc_efficiency"]) or int(epc_potential_sap) != int(
|
|
x["potential_epc_efficiency"]
|
|
):
|
|
continue
|
|
|
|
heating_kwh = extract_kwh_value(x["heating_text"])
|
|
hot_water_kwh = extract_kwh_value(x["hot_water_text"])
|
|
internal_dataset.append(
|
|
{
|
|
**epc_data,
|
|
"heating_kwh": heating_kwh,
|
|
"hot_water_kwh": hot_water_kwh,
|
|
"dataset_run_date": dataset_run_date
|
|
}
|
|
)
|
|
|
|
complete_data.extend(internal_dataset)
|
|
|
|
df = pd.DataFrame(complete_data)
|
|
# Because we collate multiple runs into a single data source, it's possible that we have duplicated data at
|
|
# the uprn level, so we dedupe based on the newest dataset_run_date
|
|
|
|
df = df.sort_values("dataset_run_date", ascending=False).drop_duplicates(subset="uprn", keep="first")
|
|
df = df.drop(columns=["dataset_run_date"])
|
|
|
|
for col in PROBLEMATIC_COLUMNS:
|
|
df[col] = df[col].astype(str)
|
|
|
|
# Save the data back to s3, but this time as a parquet file
|
|
save_dataframe_to_s3_parquet(
|
|
bucket_name="retrofit-data-dev",
|
|
file_key=f"energy_consumption/{run_date}/energy_consumption_dataset.parquet",
|
|
df=df
|
|
)
|
|
|
|
# We also estimate the energy consumption reduction from this data, by band
|
|
df["total_consumption"] = df["heating_kwh"] + df["hot_water_kwh"]
|
|
consumption_averages = df.groupby("current-energy-efficiency")["total_consumption"].mean().reset_index()
|
|
|
|
# Save the consumption averages back to s3
|
|
save_dataframe_to_s3_parquet(
|
|
bucket_name="retrofit-data-dev",
|
|
file_key=f"energy_consumption/{run_date}/consumption_averages.parquet",
|
|
df=consumption_averages
|
|
)
|