junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Map historic EPC S3 shard to domain records 🟩

Browse source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-06-29 14:45:25 +00:00
parent 74b8ad86b1
commit a372f2995f
1 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
repositories/historic_epc/historic_epc_s3_repository.py
View file

@ -5,8 +5,12 @@ from collections.abc import Callable
import pandas as pd
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
_map_historic_epc_pandas_row_to_domain,
_sanitise_postcode,
)
from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
from utils.s3 import read_csv_gz_from_s3
from utils.s3 import parse_s3_uri, read_csv_gz_from_s3
DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc"
@ -28,4 +32,8 @@ class HistoricEpcS3Repository(HistoricEpcRepository):
self._s3_root = s3_root
def get_for_postcode(self, postcode: str) -> list[HistoricEpc]:
raise NotImplementedError
pc = _sanitise_postcode(postcode)
bucket, root_prefix = parse_s3_uri(self._s3_root)
key = f"{root_prefix.rstrip('/')}/{pc}/data.csv.gz"
df = self._read_csv_gz(bucket, key)
return [_map_historic_epc_pandas_row_to_domain(row) for _, row in df.iterrows()]