Model/etl/testing_data/birmingham_pilot.py

"""
This script will create an input csv for the recommendation engine and upload it to S3, which can be used for
testing
"""
import os

import numpy as np
import pandas as pd
from epc_api.client import EpcClient
from utils.s3 import save_csv_to_s3

FILE_SIZE = 5
EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN", None)
USER_ID = 8
PORTFOLIO_ID = 54


def app():
    # For this dataset, we want 3 properties, all hourses. A mid-terrace, and end-terrace and a semi-detached

    epc_client = EpcClient(auth_token=EPC_AUTH_TOKEN)

    # Birmingham has a Local Authority Code of E08000025

    # ~~~~~~~~~~~~~~~~~~~~
    # First example
    # ~~~~~~~~~~~~~~~~~~~~
    # Let's take an EPC D property
    example_1_reponse = epc_client.domestic.search(
        params={
            "local-authority": "E08000025",
            "property-type": "house",
        },
        size=1000
    )
    example_1_reponse = example_1_reponse["rows"]
    # Get a property with a cavity wall
    example_1_reponse_filtered = [
        x for x in example_1_reponse if
        "cavity wall, as built, no insulation (assumed)" in x["walls-description"].lower()
    ]
    example_1_reponse_filtered = [
        x for x in example_1_reponse_filtered if "pitched, no insulation (assumed)" in x["roof-description"].lower()
    ]
    print(example_1_reponse_filtered[0]["postcode"])
    # 21 Penshaw Grove
    print(example_1_reponse_filtered[0]["address1"])
    # B13 9NL
    print(example_1_reponse_filtered[0]["built-form"])
    # Mid-Terrace
    print(example_1_reponse_filtered[0]["current-energy-rating"])
    # 'D'

    # ~~~~~~~~~~~~~~~~~~~~
    # Second example
    # ~~~~~~~~~~~~~~~~~~~~

    # Let's take an EPC E property
    example_2_reponse = epc_client.domestic.search(
        params={
            "local-authority": "E08000025",
            "property-type": "house",
            "energy-band": "e"
        },
        size=1000
    )
    example_2_reponse = example_2_reponse["rows"]
    # Get a solid wall example
    example_2_reponse_filtered = [
        x for x in example_2_reponse if
        "solid brick, as built, no insulation (assumed)" in x["walls-description"].lower()
    ]
    # With some existing loft insulation
    example_2_reponse_filtered = [
        x for x in example_2_reponse_filtered if "pitched, 100 mm loft insulation" in x["roof-description"].lower()
    ]

    print(example_2_reponse_filtered[0]["postcode"])
    # B13 9BY
    print(example_2_reponse_filtered[0]["address1"])
    # 2 Bloomfield Road
    print(example_2_reponse_filtered[0]["built-form"])
    # Semi-Detached
    print(example_2_reponse_filtered[0]["current-energy-rating"])
    # E

    # ~~~~~~~~~~~~~~~~~~~~
    # Third example
    # ~~~~~~~~~~~~~~~~~~~~
    example_3_reponse = epc_client.domestic.search(
        params={
            "local-authority": "E08000025",
            "property-type": "house",
            "energy-band": "f"
        },
        size=1000
    )
    example_3_reponse = example_3_reponse["rows"]
    print(example_3_reponse[2]["walls-description"])
    print(example_3_reponse[2]["floor-description"])
    print(example_3_reponse[3]["roof-description"])
    print(example_3_reponse[3]["postcode"])
    # B32 3DG
    print(example_3_reponse[3]["address1"])
    # 18 Parkside
    print(example_3_reponse[3]["built-form"])
    # End-Terrace

    # ~~~~~~~~~~~~~~~~~~~~
    # Final example
    # ~~~~~~~~~~~~~~~~~~~~
    # Let's take a flat that is a D
    example_4_reponse = epc_client.domestic.search(
        params={
            "local-authority": "E08000025",
            "property-type": "flat",
            "energy-band": "d"
        },
        size=1000
    )
    example_4_reponse = example_4_reponse["rows"]

    example_4_reponse_filtered = [
        x for x in example_4_reponse if
        "cavity wall, as built, no insulation (assumed)" in x["walls-description"].lower()
    ]
    print(example_4_reponse_filtered[0]["postcode"])
    # B14 6PU
    print(example_4_reponse_filtered[0]["address1"])
    # Flat 3

    print(example_4_reponse_filtered[0]["floor-description"])
    print(example_4_reponse_filtered[0]["property-type"])
    # Flat

    test_file = pd.DataFrame(
        [
            {"address": "21 Penshaw Grove", "postcode": "B13 9NL", "Notes": None},
            {"address": "2 Bloomfield Road", "postcode": "B13 9BY", "Notes": None},
            {"address": "18 Parkside", "postcode": "B32 3DG", "Notes": None},
            {"address": "Flat 3", "postcode": "B14 6PU", "Notes": None},
        ]
    )

    # Store the data in s3
    filename = f"{USER_ID}/{PORTFOLIO_ID}/test_inputs.csv"
    save_csv_to_s3(
        dataframe=test_file,
        bucket_name="retrofit-plan-inputs-dev",
        file_name=filename
    )

    body = {
        "portfolio_id": str(PORTFOLIO_ID),
        "housing_type": "Social",
        "goal": "Increase EPC",
        "goal_value": "C",
        "trigger_file_path": filename
    }
    print(body)