From 8171360881f957a0c1fe7e43bcd06f09d9bd7c75 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 16 Sep 2025 16:15:28 +0000 Subject: [PATCH] added month end change --- .../lambda/walthamforest_etl/docker/app.py | 132 ++++++++++++++---- etl/month_end_automation_wave_3_layout.py | 6 +- ...onth_end_automation_wave_accent_housing.py | 49 ++++++- 3 files changed, 152 insertions(+), 35 deletions(-) diff --git a/deployment/lambda/walthamforest_etl/docker/app.py b/deployment/lambda/walthamforest_etl/docker/app.py index 1cb261f..535ddd0 100644 --- a/deployment/lambda/walthamforest_etl/docker/app.py +++ b/deployment/lambda/walthamforest_etl/docker/app.py @@ -1,12 +1,13 @@ import pandas as pd import json from pprint import pprint +import os +import copy +from collections import defaultdict +from typing import List, Dict, Any, Union, Optional -def handler(event, context): - # read data for houses only - print("waltham forest set up correctly") - return None - df = pd.read_excel("../../home/Downloads/data.xlsx", sheet_name="Houses Asset Data") +def process_complex(sheet_name, group_key="ADDRESS"): + df = pd.read_excel("../../../../../home/Downloads/data.xlsx", sheet_name=sheet_name) element_cols = [ "ELEMENT GROUP", "ELEMENT CODE", "ELEMENT CODE DESCRIPTION", @@ -17,34 +18,107 @@ def handler(event, context): ] property_cols = [ - "PROP REF", "Domna", "ADDRESS", "OWNERSHIP", + "PROP REF", "ADDRESS", "OWNERSHIP", "PROP STATUS", "PROP TYPE", "PROP SUB TYPE" ] - # Group by ADDRESS (and other identifiers if needed) - result = ( - df.groupby(["ADDRESS"]) - .apply(lambda g: { - "property_info": g[property_cols].drop_duplicates().iloc[0].to_dict(), - "elements_info": [ - { - "ELEMENT GROUP": eg_name, - "elements": eg_df.drop(columns=["ELEMENT GROUP"]).to_dict(orient="records") - } - for eg_name, eg_df in g[element_cols].groupby("ELEMENT GROUP") - ] - }) - .reset_index() - .rename(columns={0: "data"}) - ) - - # Convert to list of dicts + # Prepare output records = [] - for _, row in result.iterrows(): + + # Loop through unique values in group_key (ADDRESS or BLOCK_CODE) + for val in df[group_key].unique(): + g = df[df[group_key] == val] # subset + + property_info = g[property_cols].drop_duplicates().iloc[0].to_dict() + + # build elements dict keyed by ELEMENT CODE DESCRIPTION + elements_dict = {} + for _, row in g[element_cols].drop_duplicates().iterrows(): + key = row["ELEMENT CODE DESCRIPTION"] # could also use "ELEMENT CODE" + elements_dict[key] = row.to_dict() + records.append({ - "ADDRESS": row["ADDRESS"], - **row["data"] + group_key: val, + "property_info": property_info, + "elements": elements_dict }) - json_output = json.dumps(records, ensure_ascii=False, default=str) - pprint(json_output) + return records + +def process_simple(sheet_name): + df = pd.read_excel("../../../../../home/Downloads/data.xlsx", sheet_name=sheet_name) + + records = [] + + for address in df["Address"].unique(): + g = df[df["Address"] == address].drop_duplicates() # subset for that address + row = g.iloc[0] # take first row if multiple + + # build dict of all columns except Address + elements_dict = row.drop(labels=["Address"]).to_dict() + + records.append({ + "ADDRESS": address, + "to_add": elements_dict + }) + + return records + + +def combine_records_by_address( + asset_records: List[Dict[str, Any]], + simple_records: List[Dict[str, Any]], + dest_key: str = "to_add", + unique_identifier="Address" +) -> List[Dict[str, Any]]: + """ + Merge process_house_asset_data() and process_simple() results by ADDRESS. + All columns from simple_records['to_add'] will be merged under dest_key. + """ + # Index inputs by ADDRESS + asset_by_addr = {r["ADDRESS"]: r for r in asset_records} + simple_by_addr = {r["ADDRESS"]: r for r in simple_records} + + merged: List[Dict[str, Any]] = [] + + # Use union of addresses from both sources + all_addresses = set(asset_by_addr) | set(simple_by_addr) + + for addr in sorted(all_addresses): + base = copy.deepcopy(asset_by_addr.get(addr, {"ADDRESS": addr})) + simple = simple_by_addr.get(addr) + + if simple: + base[dest_key] = simple.get("to_add", {}) + + merged.append(base) + + return merged + +def combine_records_for_flats(assets: dict, simple: list) -> dict: + """Attach BLOCK_INFO (from simple[0]) to each asset in assets.""" + if not simple or not isinstance(simple[0], dict): + return assets # nothing to add + + block_info = simple[0] + + for record in assets: + # Make sure record is a dict + record.update({"BLOCK_INFO": block_info}) + + return assets + +def handler(event, context): + # read data for houses only + assets = process_complex("Houses Asset Data") + simple = process_simple("Houses") + houses = combine_records_by_address(assets, simple, dest_key="EPC_DATA") + + # read data for flats + assets = process_complex("Chingford Rd 236-256 Properties") + simple = process_complex("CHINGFORD ROAD 236-254 Asset Bl", "BLOCK_CODE") + flats = combine_records_for_flats(assets, simple) + + + + diff --git a/etl/month_end_automation_wave_3_layout.py b/etl/month_end_automation_wave_3_layout.py index 8b18355..74038ad 100644 --- a/etl/month_end_automation_wave_3_layout.py +++ b/etl/month_end_automation_wave_3_layout.py @@ -256,17 +256,17 @@ for board, all_records in board_to_record.items(): filtered_dfs.append(design2) # Design repetitive simple - design3 = get_df(design, "design invoice type", ["archetype (simple)"], "Design Archetype repetitive") + design3 = get_df(design, "design invoice type", ["repetitive (simple)"], "Design repetitive simple") if not design1.empty: filtered_dfs.append(design3) # Design repetitive complex - design4 = get_df(design, "design invoice type", ["archetype (complex)"], "Design Archetype complex") + design4 = get_df(design, "design invoice type", ["repetitive (complex)"], "Design Repetitive complex") if not design1.empty: filtered_dfs.append(design4) # Design not specified - all_filtered = pd.concat([design1, design2, design3, design4], ignore_index=True) + all_filtered = pd.concat([df for df in (design1, design2, design3, design4) if not df.empty]) design_remaining = design.loc[~design.index.isin(all_filtered.index)] if not design_remaining.empty: design_remaining["job_type"] = "design type not specified" diff --git a/etl/month_end_automation_wave_accent_housing.py b/etl/month_end_automation_wave_accent_housing.py index b13d155..8e04c38 100644 --- a/etl/month_end_automation_wave_accent_housing.py +++ b/etl/month_end_automation_wave_accent_housing.py @@ -15,15 +15,21 @@ board_ids = [ ] empty = "Rate card info missing" - +junte = "ask junte to update" rate_card_data_2502_accent_housing = { "job_type": [ - "First half of MTP", "Second half of MTP", "Full MTP" + "First half of MTP", "Second half of MTP", "Full MTP", "Design Archetype Complex", + "Design Archetype Simple", "Design Repetitive Complex", "Design Repetitive Simple", + "Design Revision", "design type not specified", + ], "rate": [ - 150, 130, 280 + 150, 130, 280, junte, junte, junte, junte, junte, "please ask andreas" ] } +# ToDO +# Design Revision +# Design Check with Andreas rate_card_df = pd.DataFrame(rate_card_data_2502_accent_housing) @@ -91,6 +97,43 @@ full_cost = get_df(df, "mtp invoicing status", ["(v1) full cost mtp to invoice ( if not full_cost.empty: filtered_dfs.append(full_cost) +# Design archetype complex +design = get_df(df, "design invoicing status", ["to invoice"]) +design1 = get_df(design, "design invoice type", ["archetype (complex)"], "Design Archetype Complex") +if not design1.empty : + filtered_dfs.append(design1) + +# Design archetype simple +design2 = get_df(design, "design invoice type", ["archetype (simple)"], "Design Archetype Simple") +if not design1.empty: + filtered_dfs.append(design2) + +# Design repetitive simple +design3 = get_df(design, "design invoice type", ["repetitive (simple)"], "Design repetitive simple") +if not design1.empty: + filtered_dfs.append(design3) + +# Design repetitive complex +design4 = get_df(design, "design invoice type", ["repetitive (complex)"], "Design repetitive complex") +if not design1.empty: + filtered_dfs.append(design4) + +# Design not specified +all_filtered = pd.concat([df for df in (design1, design2, design3, design4) if not df.empty]) +design_remaining = design.loc[~design.index.isin(all_filtered.index)] + +if not design_remaining.empty: + design_remaining["job_type"] = "design type not specified" + filtered_dfs.append(design_remaining) + +# Design Revision +revision_letter = ['a', 'b', 'c', 'd'] +for letter in revision_letter: + design = get_df(df, "design revision invoice", [f"rev. {letter} to invoice"], "Design Revision") + if not design.empty: + filtered_dfs.append(design) + + final_df = pd.concat(filtered_dfs).reset_index(drop=True) final_df["job_type"] = final_df["job_type"].str.lower()