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()
    ]
    # Get a social housing property
    example_1_reponse_filtered = [
        x for x in example_1_reponse_filtered if x["tenure"] == "Rented (social)"
    ]

    print(example_1_reponse_filtered[0]["postcode"])
    # B13 9LT
    print(example_1_reponse_filtered[0]["address1"])
    # 113 Tenby Road
    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()
    ]
    # Get a social housing property
    example_2_reponse_filtered = [
        x for x in example_2_reponse_filtered if x["tenure"] == "Rented (social)"
    ]

    print(example_2_reponse_filtered[0]["postcode"])
    # B28 8JF
    print(example_2_reponse_filtered[0]["address1"])
    # 139 School 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"]
    # Get a social housing property]
    example_3_reponse_filtered = [
        x for x in example_3_reponse if x["tenure"] == "Rented (social)"
    ]

    print(example_3_reponse_filtered[4]["walls-description"])
    print(example_3_reponse_filtered[4]["floor-description"])
    print(example_3_reponse_filtered[4]["roof-description"])
    print(example_3_reponse_filtered[4]["postcode"])
    # B32 1SL
    print(example_3_reponse_filtered[4]["address1"])
    # 77 Simmons Drive
    print(example_3_reponse_filtered[4]["built-form"])
    # Semi-Detached

    # ~~~~~~~~~~~~~~~~~~~~
    # 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()
    ]
    # Get a social housing property
    example_4_reponse_filtered = [
        x for x in example_4_reponse_filtered if x["tenure"] == "Rented (social)"
    ]
    print(example_4_reponse_filtered[0]["postcode"])
    # B32 1LS
    print(example_4_reponse_filtered[0]["address1"])
    # Flat 2

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

    test_file = pd.DataFrame(
        [
            # New properties
            {"address": "113 Tenby Road", "postcode": "B13 9LT", "Notes": None},
            {"address": "139 School Road", "postcode": "B28 8JF", "Notes": None},
            {"address": "77 Simmons Drive", "postcode": "B32 1SL", "Notes": None},
            {"address": "Flat 2, 54 Wedgewood Road", "postcode": "B32 1LS", "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)