From 8b2f74766eb7fa7108609c9520a65e2e08f18f54 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Fri, 17 Jul 2026 12:16:19 +0100 Subject: [PATCH 1/6] Copy existing electrcity_sharepoint pipeline ref #321 --- .../electricity_sharepoint.py | 285 ++++++++++++++++++ 1 file changed, 285 insertions(+) create mode 100644 elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py diff --git a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py new file mode 100644 index 00000000..f9369d8a --- /dev/null +++ b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py @@ -0,0 +1,285 @@ +# /// script +# requires-python = ">=3.13" +# dependencies = [ +# "pandas>=3", +# "elt-common[m365]", +# "python-calamine", +# ] +# +# [tool.uv.sources] +# elt-common = { path = "../../../../../elt-common" } +# /// +import concurrent.futures +import io +import os +import pathlib +from typing import Iterator, Sequence + +import dlt +import dlt.common.logger as logger +import pandas as pd +import pendulum +import pyarrow.compute as pc +from dlt.sources import TDataItems +from elt_common.cli_utils import cli_main +from elt_common.dlt_destinations.pyiceberg.helpers import load_iceberg_table +from elt_common.dlt_destinations.pyiceberg.pyiceberg_adapter import ( + PartitionTrBuilder, + pyiceberg_adapter, +) +from elt_common.dlt_sources.m365 import ( + M365DriveItem, + sharepoint, +) + +CSV_PREAMBLE_ANCHOR = "time" +COL_DATE_TIME = "date_time" +COL_TOTAL_POWER = "isis_elec_total_power_mw" +EXCEL_ENGINE = "calamine" +EXCEL_SKIP_ROWS = 7 +MAX_WORKERS = min(8, (os.cpu_count() or 1) + 4) +ONE_DAY_SECS = 24 * 60 * 60 +PIPELINE_NAME = "electricity_sharepoint" +RDM_TIMEZONE = "Europe/London" + +# Source details +# +# There are 3 known formats for the data. Each contains a preamble of the form: +# +# Site Information: +# RAL ISIS RDM +# RAL ISIS RDM +# Controller: ISIS +# Controller description: ISIS Energy Totals +# Status: Online +# +# This preamble is discarded from all formats. +# +# 1. Excel (old, manual export format) +# ------------------------------------ +# +# These were produced by manual export from the original system. +# +# There are two columns: "Time,Total Power (MW)". Time is in the format "YYYY-mm-DDTHH:MM:SS". +# +# 2. Automated CSV export +# ----------------------- +# +# An automation deposits .csv files at regular intervals. +# +# There are three columns: "Time,Date,ISIS Elec Total Power", where Time is in format HH:MM:SS and Date is in format DD/mm/YY. +# Several of these files can be concatenated together to form larger timespan records - they are concatenated as is and repeated +# preamble sections are not discarded. The variable CSV_PREAMBLE_ANCHOR defines the line at which a new section occurs. +# +# 3. Manual CSV export +# -------------------- +# +# These were produced more recently by manual export from the original system. +# There are three columns: "Time,ISIS Elec Total Energy,ISIS Elec Total Power", where Time is in format "DD/mm/YY HH:MM:SS". +# The second column is discarded. + + +def to_utc(ts: pd.Series) -> pd.Series: + """Assumes timezone unaware data and converts to UTC""" + return ts.dt.tz_localize(RDM_TIMEZONE).dt.tz_convert("UTC") + + +def csv_section_to_df(file_name: str, lines: Sequence[str]) -> pd.DataFrame | None: + """Parse csv and return a DataFrame if the times are valid. None if not""" + df_raw = pd.read_csv(io.StringIO("\n".join(lines))) + # clean up column name (strip any whitespace) + df_raw.columns = df_raw.columns.str.strip() + cols = [c for c in df_raw.columns] + assert len(cols) == 3 + try: + if cols[1].strip() == "Date": + # Automated CSV format + df = to_utc( + pd.to_datetime( + df_raw["Date"] + " " + df_raw["Time"], format="%d/%m/%y %H:%M:%S" + ) # type: ignore + ).to_frame(name=COL_DATE_TIME) + else: + # Manual CSV format + df = to_utc( + pd.to_datetime(df_raw["Time"], format="%d/%m/%y %H:%M:%S") + ).to_frame(name=COL_DATE_TIME) + except ValueError as exc: + # Pandas 3 uses ValueError for conversions that produce ambiguous/non-existent times + msg = str(exc) + if "ambiguous" in msg or "nonexistent" in msg: + logger.warning( + f"'Error loading section of {file_name}'. DST issues detected: {str(exc)}" + ) + return None + else: + raise + + assert "power" in cols[2].lower() + df[COL_TOTAL_POWER] = df_raw[cols[2]] + return df + + +def read_power_consumption_csv( + file_content: io.BytesIO, + file_name: str, +) -> pd.DataFrame | None: + # See comment at the top of this describing the format + + def _append_if_not_none(seq, df): + if df is not None: + seq.append(df) + + metadata_anchor = "site information" + sections, current_lines, in_data = [], [], False + for line in file_content.getvalue().decode().splitlines(): + line = line.strip() + line_lower = line.lower() + + if line_lower.startswith(CSV_PREAMBLE_ANCHOR): + # Save any previous section + if current_lines: + _append_if_not_none( + sections, csv_section_to_df(file_name, current_lines) + ) + # start new section with header + current_lines = [line] + in_data = True + + elif in_data: + if line_lower.startswith(metadata_anchor): + # Metadata block + in_data = False + continue + else: + current_lines.append(line) + + # the last section + if current_lines: + _append_if_not_none(sections, csv_section_to_df(file_name, current_lines)) + + # concatenate + if sections: + return pd.concat(sections, ignore_index=True) + + return None + + +def read_power_consumption_excel(file_content: io.BytesIO) -> pd.DataFrame: + # See comment at the top of this describing the format + df_raw = pd.read_excel(file_content, engine=EXCEL_ENGINE, skiprows=EXCEL_SKIP_ROWS) + df_raw = df_raw.rename(columns={"Time": COL_DATE_TIME}) + df_raw[COL_DATE_TIME] = to_utc(df_raw[COL_DATE_TIME]) # type: ignore + return df_raw + + +@dlt.transformer(section="m365") +def extract_content_and_read(items: Iterator[M365DriveItem]) -> Iterator[TDataItems]: + """Extracts the file content and reads it assuming it is a .csv or a .xlsx file + + Combines the Date and Time columns into a DateTime field to simplify incremental + processing. + + :param items: An iterator of dicts describing the file content + :param max_workers (optional): How many threads to use to process the files. + Defaults to a maximum defined by concurrent.futures.ThreadPoolExecutor + """ + + # The files are all independent. Process them in parallel and combine for a single yield + def read_as_dataframe(file_obj: M365DriveItem) -> pd.DataFrame | None: + file_name = file_obj["file_name"] + file_bytes = file_obj.read_bytes() + file_content = io.BytesIO(file_bytes) + logger.debug(f"Filename '{file_name}' has size {len(file_bytes)} bytes.") + match pathlib.Path(file_name).suffix: + case ".csv": + df = read_power_consumption_csv(file_content, file_name) + case ".xlsx": + df = read_power_consumption_excel(file_content) + case _: + raise RuntimeError(f"Unsupported file extension in '{file_name}'") + + if df is not None: + df["file_name"] = file_name + return df + + df_batch = None + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + future_to_file_item = { + executor.submit(read_as_dataframe, file_obj): file_obj for file_obj in items + } + for future in concurrent.futures.as_completed(future_to_file_item): + df = future.result() + if df is not None: + df_batch = pd.concat((df_batch, df)) if df_batch is not None else df + + yield df_batch + + +@dlt.resource( + primary_key="DateTime", + write_disposition={"disposition": "merge", "strategy": "upsert"}, +) +def rdm_data( + site_url: str = dlt.config.value, + root_dir: str = dlt.config.value, + backfill: bool = False, + backfill_glob: str | None = None, +) -> Iterator[TDataItems]: + if backfill: + if backfill_glob is not None: + file_globs = [backfill_glob] + else: + file_globs = [ + f"{root_dir}/**/*.xlsx", + f"{root_dir}/**/*-daily.csv", + f"{root_dir}/**/*-manual-export.csv", + ] + modified_after = None + else: + file_globs = [f"{root_dir}/*-ISIS.csv"] + modified_after = get_latest_timestamp(dlt.current.pipeline()) + + for file_glob in file_globs: + file_listing = sharepoint( + site_url=site_url, + file_glob=file_glob, + extract_content=False, + modified_after=modified_after, + ) + reader = file_listing | extract_content_and_read() + yield from reader + + +def get_latest_timestamp(pipeline: dlt.Pipeline) -> pendulum.DateTime | None: + """Retrieve the timestamp loaded into the warehouse""" + existing_rdm_data = load_iceberg_table(pipeline, "rdm_data") + if existing_rdm_data is None: + return None + + logger.debug("Destination table exists, finding latest timestamp.") + c_datetime = "date_time" + scan = existing_rdm_data.scan(selected_fields=(c_datetime,)) + running_max = None + for batch in scan.to_arrow_batch_reader(): + batch_max = pc.max(batch[c_datetime]) # type: ignore + if batch_max.is_valid: + running_max = ( + batch_max if running_max is None else pc.max([running_max, batch_max]) # type: ignore + ) + + latest_ts = ( + pendulum.instance(running_max.as_py()) if running_max is not None else None + ) + logger.debug(f"Latest record has timestamp: {latest_ts}") + return latest_ts + + +if __name__ == "__main__": + cli_main( + pipeline_name=PIPELINE_NAME, + data_generator=pyiceberg_adapter( + rdm_data, partition=PartitionTrBuilder.year("date_time") + ), + source_domain="estates", + ) From 3a3985ff8a89c9d0a99cfa6623d8b1b5a0f290c0 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Fri, 17 Jul 2026 16:06:41 +0100 Subject: [PATCH 2/6] feat(elt-pipelines): Port electricity_sharepoint to elt ref #321 --- .../electricity_sharepoint.py | 335 +++++------------- .../estates/electricity_sharepoint/parsing.py | 149 ++++++++ elt-pipelines/pyproject.toml | 6 + elt-pipelines/uv.lock | 10 +- 4 files changed, 258 insertions(+), 242 deletions(-) create mode 100644 elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py diff --git a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py index f9369d8a..d7c8ede1 100644 --- a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py +++ b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py @@ -1,199 +1,125 @@ -# /// script -# requires-python = ">=3.13" -# dependencies = [ -# "pandas>=3", -# "elt-common[m365]", -# "python-calamine", -# ] -# -# [tool.uv.sources] -# elt-common = { path = "../../../../../elt-common" } -# /// +"""Extracts measurements of ISIS's electricity usage from the ISIS Sustainability Sharepoint site""" + import concurrent.futures +import datetime as dt import io import os +import logging import pathlib -from typing import Iterator, Sequence -import dlt -import dlt.common.logger as logger import pandas as pd -import pendulum -import pyarrow.compute as pc -from dlt.sources import TDataItems -from elt_common.cli_utils import cli_main -from elt_common.dlt_destinations.pyiceberg.helpers import load_iceberg_table -from elt_common.dlt_destinations.pyiceberg.pyiceberg_adapter import ( - PartitionTrBuilder, - pyiceberg_adapter, -) -from elt_common.dlt_sources.m365 import ( - M365DriveItem, - sharepoint, +import pyarrow as pa + +from parsing import read_power_consumption_csv, read_power_consumption_excel + +from elt_common.extract import ( + BaseExtract, + ResourceProperties, + Watermark, + ResourceWriteProperties, ) +from elt_common.sources.m365.client import SPListClient, M365File +from elt_common.sources.m365.credentials import M365Credentials -CSV_PREAMBLE_ANCHOR = "time" -COL_DATE_TIME = "date_time" -COL_TOTAL_POWER = "isis_elec_total_power_mw" -EXCEL_ENGINE = "calamine" -EXCEL_SKIP_ROWS = 7 +SITE_URL = "https://stfc365.sharepoint.com/sites/ISISSustainability" MAX_WORKERS = min(8, (os.cpu_count() or 1) + 4) -ONE_DAY_SECS = 24 * 60 * 60 -PIPELINE_NAME = "electricity_sharepoint" -RDM_TIMEZONE = "Europe/London" -# Source details -# -# There are 3 known formats for the data. Each contains a preamble of the form: -# -# Site Information: -# RAL ISIS RDM -# RAL ISIS RDM -# Controller: ISIS -# Controller description: ISIS Energy Totals -# Status: Online -# -# This preamble is discarded from all formats. -# -# 1. Excel (old, manual export format) -# ------------------------------------ -# -# These were produced by manual export from the original system. -# -# There are two columns: "Time,Total Power (MW)". Time is in the format "YYYY-mm-DDTHH:MM:SS". -# -# 2. Automated CSV export -# ----------------------- -# -# An automation deposits .csv files at regular intervals. -# -# There are three columns: "Time,Date,ISIS Elec Total Power", where Time is in format HH:MM:SS and Date is in format DD/mm/YY. -# Several of these files can be concatenated together to form larger timespan records - they are concatenated as is and repeated -# preamble sections are not discarded. The variable CSV_PREAMBLE_ANCHOR defines the line at which a new section occurs. -# -# 3. Manual CSV export -# -------------------- -# -# These were produced more recently by manual export from the original system. -# There are three columns: "Time,ISIS Elec Total Energy,ISIS Elec Total Power", where Time is in format "DD/mm/YY HH:MM:SS". -# The second column is discarded. +LOGGER = logging.getLogger(__name__) +_root_path = "/General/RDM Data" +_default_backfill_globs = [ + "**/*.xlsx", + "**/*-daily.csv", + "**/*-manual-export.csv", +] -def to_utc(ts: pd.Series) -> pd.Series: - """Assumes timezone unaware data and converts to UTC""" - return ts.dt.tz_localize(RDM_TIMEZONE).dt.tz_convert("UTC") +class Configuration(M365Credentials): + backfill: bool = False + backfill_globs: list[str] = [] -def csv_section_to_df(file_name: str, lines: Sequence[str]) -> pd.DataFrame | None: - """Parse csv and return a DataFrame if the times are valid. None if not""" - df_raw = pd.read_csv(io.StringIO("\n".join(lines))) - # clean up column name (strip any whitespace) - df_raw.columns = df_raw.columns.str.strip() - cols = [c for c in df_raw.columns] - assert len(cols) == 3 - try: - if cols[1].strip() == "Date": - # Automated CSV format - df = to_utc( - pd.to_datetime( - df_raw["Date"] + " " + df_raw["Time"], format="%d/%m/%y %H:%M:%S" - ) # type: ignore - ).to_frame(name=COL_DATE_TIME) - else: - # Manual CSV format - df = to_utc( - pd.to_datetime(df_raw["Time"], format="%d/%m/%y %H:%M:%S") - ).to_frame(name=COL_DATE_TIME) - except ValueError as exc: - # Pandas 3 uses ValueError for conversions that produce ambiguous/non-existent times - msg = str(exc) - if "ambiguous" in msg or "nonexistent" in msg: - logger.warning( - f"'Error loading section of {file_name}'. DST issues detected: {str(exc)}" - ) - return None - else: - raise + @property + def glob_patterns(self): + if not self.backfill: + return ["*-ISIS.csv"] + return self.backfill_globs if self.backfill_globs else _default_backfill_globs - assert "power" in cols[2].lower() - df[COL_TOTAL_POWER] = df_raw[cols[2]] - return df +class Extract(BaseExtract): + config_cls = Configuration -def read_power_consumption_csv( - file_content: io.BytesIO, - file_name: str, -) -> pd.DataFrame | None: - # See comment at the top of this describing the format + def __init__(self, cfg: Configuration): + super().__init__(cfg) + self._client = SPListClient(SITE_URL, cfg) + self._backfilling = cfg.backfill + self._glob_patterns = cfg.glob_patterns - def _append_if_not_none(seq, df): - if df is not None: - seq.append(df) + LOGGER.debug(f"Searching for files matching: {self._glob_patterns}") - metadata_anchor = "site information" - sections, current_lines, in_data = [], [], False - for line in file_content.getvalue().decode().splitlines(): - line = line.strip() - line_lower = line.lower() + def extract_resource_properties(self): + yield ( + "rdm_data", + ResourceProperties( + extractor=self._extract_electricity_usage, + write_properties=ResourceWriteProperties( + merge_on=["date_time"], write_mode="merge" + ), + watermark_column="date_time", + ), + ) - if line_lower.startswith(CSV_PREAMBLE_ANCHOR): - # Save any previous section - if current_lines: - _append_if_not_none( - sections, csv_section_to_df(file_name, current_lines) + def _extract_electricity_usage(self, w: Watermark | None): + watermark_value: dt.datetime | None = None + if w and self._backfilling: + LOGGER.debug("Ignoring watermark because this is a backfill") + elif w: + if not isinstance(w.value, dt.datetime): + LOGGER.warning( + f"Ignoring watermark for electricity SP because it was '{w.value}', not a datetime" ) - # start new section with header - current_lines = [line] - in_data = True - - elif in_data: - if line_lower.startswith(metadata_anchor): - # Metadata block - in_data = False - continue else: - current_lines.append(line) - - # the last section - if current_lines: - _append_if_not_none(sections, csv_section_to_df(file_name, current_lines)) - - # concatenate - if sections: - return pd.concat(sections, ignore_index=True) + watermark_value = w.value + LOGGER.debug(f"Only fetching files modified after {watermark_value}") + + files = [] + for pattern in self._glob_patterns: + files.extend( + self._client.glob( + _root_path, pattern=pattern, modified_after=watermark_value + ) + ) - return None + LOGGER.debug(f"Matched {len(files)} files") + resolved = self._read_files(files) + df = read_contents_to_dataframe(resolved) + yield pa.Table.from_pandas(df, preserve_index=False) -def read_power_consumption_excel(file_content: io.BytesIO) -> pd.DataFrame: - # See comment at the top of this describing the format - df_raw = pd.read_excel(file_content, engine=EXCEL_ENGINE, skiprows=EXCEL_SKIP_ROWS) - df_raw = df_raw.rename(columns={"Time": COL_DATE_TIME}) - df_raw[COL_DATE_TIME] = to_utc(df_raw[COL_DATE_TIME]) # type: ignore - return df_raw + def _read_files(self, files: list[M365File]) -> list[tuple[str, bytes]]: + results = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + future_to_file_item = { + executor.submit(self._client.read_file, f.path): f for f in files + } + for future in concurrent.futures.as_completed(future_to_file_item): + file = future_to_file_item[future] + results.append((file.name, future.result())) + return results -@dlt.transformer(section="m365") -def extract_content_and_read(items: Iterator[M365DriveItem]) -> Iterator[TDataItems]: - """Extracts the file content and reads it assuming it is a .csv or a .xlsx file - Combines the Date and Time columns into a DateTime field to simplify incremental - processing. +def read_contents_to_dataframe(files: list[tuple[str, bytes]]): + """Extracts data from files into a single, combined dataframe - :param items: An iterator of dicts describing the file content - :param max_workers (optional): How many threads to use to process the files. - Defaults to a maximum defined by concurrent.futures.ThreadPoolExecutor + :param files: (name, content) pairs for the files to extract data from """ - # The files are all independent. Process them in parallel and combine for a single yield - def read_as_dataframe(file_obj: M365DriveItem) -> pd.DataFrame | None: - file_name = file_obj["file_name"] - file_bytes = file_obj.read_bytes() + def read_as_dataframe(file_name, file_bytes) -> pd.DataFrame | None: file_content = io.BytesIO(file_bytes) - logger.debug(f"Filename '{file_name}' has size {len(file_bytes)} bytes.") + LOGGER.debug(f"File '{file_name}' has size {len(file_bytes)} bytes.") match pathlib.Path(file_name).suffix: case ".csv": - df = read_power_consumption_csv(file_content, file_name) + df = read_power_consumption_csv(file_name, file_content) case ".xlsx": df = read_power_consumption_excel(file_content) case _: @@ -203,83 +129,10 @@ def read_as_dataframe(file_obj: M365DriveItem) -> pd.DataFrame | None: df["file_name"] = file_name return df - df_batch = None - with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: - future_to_file_item = { - executor.submit(read_as_dataframe, file_obj): file_obj for file_obj in items - } - for future in concurrent.futures.as_completed(future_to_file_item): - df = future.result() - if df is not None: - df_batch = pd.concat((df_batch, df)) if df_batch is not None else df - - yield df_batch - - -@dlt.resource( - primary_key="DateTime", - write_disposition={"disposition": "merge", "strategy": "upsert"}, -) -def rdm_data( - site_url: str = dlt.config.value, - root_dir: str = dlt.config.value, - backfill: bool = False, - backfill_glob: str | None = None, -) -> Iterator[TDataItems]: - if backfill: - if backfill_glob is not None: - file_globs = [backfill_glob] - else: - file_globs = [ - f"{root_dir}/**/*.xlsx", - f"{root_dir}/**/*-daily.csv", - f"{root_dir}/**/*-manual-export.csv", - ] - modified_after = None - else: - file_globs = [f"{root_dir}/*-ISIS.csv"] - modified_after = get_latest_timestamp(dlt.current.pipeline()) - - for file_glob in file_globs: - file_listing = sharepoint( - site_url=site_url, - file_glob=file_glob, - extract_content=False, - modified_after=modified_after, - ) - reader = file_listing | extract_content_and_read() - yield from reader - - -def get_latest_timestamp(pipeline: dlt.Pipeline) -> pendulum.DateTime | None: - """Retrieve the timestamp loaded into the warehouse""" - existing_rdm_data = load_iceberg_table(pipeline, "rdm_data") - if existing_rdm_data is None: - return None - - logger.debug("Destination table exists, finding latest timestamp.") - c_datetime = "date_time" - scan = existing_rdm_data.scan(selected_fields=(c_datetime,)) - running_max = None - for batch in scan.to_arrow_batch_reader(): - batch_max = pc.max(batch[c_datetime]) # type: ignore - if batch_max.is_valid: - running_max = ( - batch_max if running_max is None else pc.max([running_max, batch_max]) # type: ignore - ) - - latest_ts = ( - pendulum.instance(running_max.as_py()) if running_max is not None else None - ) - logger.debug(f"Latest record has timestamp: {latest_ts}") - return latest_ts - + df_batch: pd.DataFrame | None = None + for name, file_contents in files: + df = read_as_dataframe(name, file_contents) + if df is not None: + df_batch = df if df_batch is None else pd.concat((df_batch, df)) -if __name__ == "__main__": - cli_main( - pipeline_name=PIPELINE_NAME, - data_generator=pyiceberg_adapter( - rdm_data, partition=PartitionTrBuilder.year("date_time") - ), - source_domain="estates", - ) + return df_batch diff --git a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py new file mode 100644 index 00000000..e7f48a14 --- /dev/null +++ b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py @@ -0,0 +1,149 @@ +"""Functions for parsing data from the electricity files + +There are 3 known formats for the data. Each contains a preamble of the form: + +Site Information: +RAL ISIS RDM +RAL ISIS RDM +Controller: ISIS +Controller description: ISIS Energy Totals +Status: Online + +This preamble is discarded from all formats. + +1. Excel (old, manual export format) +------------------------------------ + +These were produced by manual export from the original system. + +There are two columns: "Time,Total Power (MW)". Time is in the format "YYYY-mm-DDTHH:MM:SS". + +2. Automated CSV export +----------------------- + +An automation deposits .csv files at regular intervals. + +There are three columns: "Time,Date,ISIS Elec Total Power", where Time is in format HH:MM:SS and Date is in format DD/mm/YY. +Several of these files can be concatenated together to form larger timespan records - they are concatenated as is and repeated +preamble sections are not discarded. The variable CSV_PREAMBLE_ANCHOR defines the line at which a new section occurs. + +3. Manual CSV export +-------------------- + +These were produced more recently by manual export from the original system. +There are three columns: "Time,ISIS Elec Total Energy,ISIS Elec Total Power", where Time is in format "DD/mm/YY HH:MM:SS". +The second column is discarded. +""" + +import io +import logging +import pandas as pd + +from typing import Sequence + +CSV_PREAMBLE_ANCHOR = "time" +COL_DATE_TIME = "date_time" +COL_TOTAL_POWER = "isis_elec_total_power_mw" +EXCEL_SKIP_ROWS = 7 +RDM_TIMEZONE = "Europe/London" + +LOGGER = logging.getLogger(__name__) + + +def read_power_consumption_excel(file_content: io.BytesIO) -> pd.DataFrame: + # See module comment for description of the format + df_raw = pd.read_excel(file_content, skiprows=EXCEL_SKIP_ROWS) + df_raw = df_raw.rename( + columns={"Time": COL_DATE_TIME, "ISIS Elec Total Power {MW}": COL_TOTAL_POWER} + ) + df_raw[COL_DATE_TIME] = _to_utc(df_raw[COL_DATE_TIME]) # type: ignore + return df_raw + + +def read_power_consumption_csv( + file_name: str, file_content: io.BytesIO +) -> pd.DataFrame | None: + # See module comment for description of the format + + def _append_if_not_none(seq, df): + if df is not None: + seq.append(df) + + metadata_anchor = "site information" + sections = [] + current_lines: list[str] = [] + in_data = False + for line in file_content.getvalue().decode().splitlines(): + line = line.strip() + line_lower = line.lower() + + if line_lower.startswith(CSV_PREAMBLE_ANCHOR): + # Save any previous section + if current_lines: + _append_if_not_none( + sections, _csv_section_to_df(file_name, current_lines) + ) + # start new section with header + current_lines = [line] + in_data = True + + elif in_data: + if line_lower.startswith(metadata_anchor): + # Metadata block + in_data = False + continue + else: + current_lines.append(line) + + # the last section + if current_lines: + _append_if_not_none(sections, _csv_section_to_df(file_name, current_lines)) + + # concatenate + if sections: + return pd.concat(sections) + + return None + + +def _csv_section_to_df(file_name: str, lines: Sequence[str]) -> pd.DataFrame | None: + """Parse csv and return a DataFrame if the times are valid. None if not""" + buff = io.StringIO("\n".join(lines)) + df_raw: pd.DataFrame = pd.read_csv(buff) # type: ignore + + # clean up column names (strip any whitespace) + df_raw.columns = df_raw.columns.str.strip() + cols = [c for c in df_raw.columns] + assert len(cols) == 3 + assert "power" in cols[2].lower() + + try: + if cols[1].strip() == "Date": + # Automated CSV format + raw_datetime = df_raw["Date"] + " " + df_raw["Time"] + else: + # Manual CSV format + raw_datetime = df_raw["Time"] + + df_datetime: pd.Series = pd.to_datetime( # type: ignore + raw_datetime, format="%d/%m/%y %H:%M:%S" + ) + df = _to_utc(df_datetime).to_frame(name=COL_DATE_TIME) + except ValueError as exc: + # Pandas 3 uses ValueError for conversions that produce ambiguous/non-existent times + msg = str(exc) + if "ambiguous" in msg or "nonexistent" in msg: + LOGGER.warning( + f"'Error loading section of {file_name}'. DST issues detected: {str(exc)}" + ) + return None + else: + raise + + df[COL_TOTAL_POWER] = df_raw[cols[2]] + return df + + +def _to_utc(ts: pd.Series) -> pd.Series: + """Assumes timezone unaware data and converts to UTC""" + return ts.dt.tz_localize(RDM_TIMEZONE).dt.tz_convert("UTC") diff --git a/elt-pipelines/pyproject.toml b/elt-pipelines/pyproject.toml index a0a767ea..fb6cd35f 100644 --- a/elt-pipelines/pyproject.toml +++ b/elt-pipelines/pyproject.toml @@ -21,6 +21,12 @@ accelerator_sharepoint = [ "elt-common[m365]" ] +electricity_sharepoint = [ + "pandas >= 3.0.3", + "openpyxl >= 3.1.5", + "elt-common[m365]" +] + [tool.uv.sources] elt-common = { path = "../elt-common", editable = true } diff --git a/elt-pipelines/uv.lock b/elt-pipelines/uv.lock index 2fa17f2f..c28b876a 100644 --- a/elt-pipelines/uv.lock +++ b/elt-pipelines/uv.lock @@ -526,6 +526,11 @@ accelerator-sharepoint = [ { name = "openpyxl" }, { name = "pandas" }, ] +electricity-sharepoint = [ + { name = "elt-common", extra = ["m365"] }, + { name = "openpyxl" }, + { name = "pandas" }, +] statusdisplay = [ { name = "pyarrow" }, { name = "requests" }, @@ -540,13 +545,16 @@ dev = [ requires-dist = [ { name = "elt-common", editable = "../elt-common" }, { name = "elt-common", extras = ["m365"], marker = "extra == 'accelerator-sharepoint'", editable = "../elt-common" }, + { name = "elt-common", extras = ["m365"], marker = "extra == 'electricity-sharepoint'", editable = "../elt-common" }, { name = "openpyxl", marker = "extra == 'accelerator-sharepoint'", specifier = ">=3.1.5" }, + { name = "openpyxl", marker = "extra == 'electricity-sharepoint'", specifier = ">=3.1.5" }, { name = "pandas", marker = "extra == 'accelerator-sharepoint'", specifier = ">=3.0.3" }, + { name = "pandas", marker = "extra == 'electricity-sharepoint'", specifier = ">=3.0.3" }, { name = "pyarrow", marker = "extra == 'statusdisplay'", specifier = ">=24.0.0" }, { name = "pydantic-settings", specifier = ">=2.14.2" }, { name = "requests", marker = "extra == 'statusdisplay'", specifier = ">=2.34.2" }, ] -provides-extras = ["statusdisplay", "accelerator-sharepoint"] +provides-extras = ["statusdisplay", "accelerator-sharepoint", "electricity-sharepoint"] [package.metadata.requires-dev] dev = [{ name = "prek", specifier = ">=0.4.5" }] From b6371ccf06b8813f4b4f061a5020398202cdc09a Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 23 Jul 2026 13:15:59 +0100 Subject: [PATCH 3/6] Combine optional dependencies for sharepoint pipelines ref #321 --- elt-pipelines/pyproject.toml | 8 +------- elt-pipelines/uv.lock | 18 +++++------------- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/elt-pipelines/pyproject.toml b/elt-pipelines/pyproject.toml index fb6cd35f..51e7a07f 100644 --- a/elt-pipelines/pyproject.toml +++ b/elt-pipelines/pyproject.toml @@ -15,13 +15,7 @@ statusdisplay = [ "requests>=2.34.2", ] -accelerator_sharepoint = [ - "pandas >= 3.0.3", - "openpyxl >= 3.1.5", - "elt-common[m365]" -] - -electricity_sharepoint = [ +sharepoint = [ "pandas >= 3.0.3", "openpyxl >= 3.1.5", "elt-common[m365]" diff --git a/elt-pipelines/uv.lock b/elt-pipelines/uv.lock index c28b876a..a5da87b5 100644 --- a/elt-pipelines/uv.lock +++ b/elt-pipelines/uv.lock @@ -521,12 +521,7 @@ dependencies = [ ] [package.optional-dependencies] -accelerator-sharepoint = [ - { name = "elt-common", extra = ["m365"] }, - { name = "openpyxl" }, - { name = "pandas" }, -] -electricity-sharepoint = [ +sharepoint = [ { name = "elt-common", extra = ["m365"] }, { name = "openpyxl" }, { name = "pandas" }, @@ -544,17 +539,14 @@ dev = [ [package.metadata] requires-dist = [ { name = "elt-common", editable = "../elt-common" }, - { name = "elt-common", extras = ["m365"], marker = "extra == 'accelerator-sharepoint'", editable = "../elt-common" }, - { name = "elt-common", extras = ["m365"], marker = "extra == 'electricity-sharepoint'", editable = "../elt-common" }, - { name = "openpyxl", marker = "extra == 'accelerator-sharepoint'", specifier = ">=3.1.5" }, - { name = "openpyxl", marker = "extra == 'electricity-sharepoint'", specifier = ">=3.1.5" }, - { name = "pandas", marker = "extra == 'accelerator-sharepoint'", specifier = ">=3.0.3" }, - { name = "pandas", marker = "extra == 'electricity-sharepoint'", specifier = ">=3.0.3" }, + { name = "elt-common", extras = ["m365"], marker = "extra == 'sharepoint'", editable = "../elt-common" }, + { name = "openpyxl", marker = "extra == 'sharepoint'", specifier = ">=3.1.5" }, + { name = "pandas", marker = "extra == 'sharepoint'", specifier = ">=3.0.3" }, { name = "pyarrow", marker = "extra == 'statusdisplay'", specifier = ">=24.0.0" }, { name = "pydantic-settings", specifier = ">=2.14.2" }, { name = "requests", marker = "extra == 'statusdisplay'", specifier = ">=2.34.2" }, ] -provides-extras = ["statusdisplay", "accelerator-sharepoint", "electricity-sharepoint"] +provides-extras = ["statusdisplay", "sharepoint"] [package.metadata.requires-dev] dev = [{ name = "prek", specifier = ">=0.4.5" }] From 6735ed3714fedc78257919ca8aa246f49d7d4cff Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 23 Jul 2026 15:22:56 +0100 Subject: [PATCH 4/6] Handle cases where no data is extracted --- elt-common/src/elt_common/runner.py | 4 ++++ .../estates/electricity_sharepoint/electricity_sharepoint.py | 5 ++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/elt-common/src/elt_common/runner.py b/elt-common/src/elt_common/runner.py index 7a71310d..afe41553 100644 --- a/elt-common/src/elt_common/runner.py +++ b/elt-common/src/elt_common/runner.py @@ -76,6 +76,10 @@ def run_ingest(job: ELTJobManifest) -> dict[str, int]: watermarks: list[Watermark] = [] for data in table_props.extractor(watermark_before_extract): + if not data: + LOGGER.info("No rows extracted, skipping") + continue + # 'replace' really means delete the contents of the table, then append the new data. # Extractors can return multiple chunks of data, in which case only the first chunk # should cause a deletion, whilst the remaining chunks should be appended. diff --git a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py index d7c8ede1..18cdafaf 100644 --- a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py +++ b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py @@ -93,7 +93,10 @@ def _extract_electricity_usage(self, w: Watermark | None): resolved = self._read_files(files) df = read_contents_to_dataframe(resolved) - yield pa.Table.from_pandas(df, preserve_index=False) + if df is not None and df.size > 0: + yield pa.Table.from_pandas(df, preserve_index=False) + else: + yield pa.Table.from_pylist([]) def _read_files(self, files: list[M365File]) -> list[tuple[str, bytes]]: results = [] From 9ebaee4e5063bd7fa305e711f9dab4131392a4bc Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 23 Jul 2026 15:27:22 +0100 Subject: [PATCH 5/6] Don't use assert for runtime checks --- .../ingest/estates/electricity_sharepoint/parsing.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py index e7f48a14..90790ae5 100644 --- a/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py +++ b/elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py @@ -114,8 +114,9 @@ def _csv_section_to_df(file_name: str, lines: Sequence[str]) -> pd.DataFrame | N # clean up column names (strip any whitespace) df_raw.columns = df_raw.columns.str.strip() cols = [c for c in df_raw.columns] - assert len(cols) == 3 - assert "power" in cols[2].lower() + if len(cols) != 3 or "power" not in cols[2].lower(): + LOGGER.warning(f"Columns in {file_name} are an unexpected format: {cols}") + return None try: if cols[1].strip() == "Date": From f5d171e4ac2f61aa98448156d17aaead98919bd6 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 23 Jul 2026 15:44:26 +0100 Subject: [PATCH 6/6] Test that empty table causes no write at all --- elt-common/tests/unit_tests/test_runner.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/elt-common/tests/unit_tests/test_runner.py b/elt-common/tests/unit_tests/test_runner.py index bde37c42..90d61667 100644 --- a/elt-common/tests/unit_tests/test_runner.py +++ b/elt-common/tests/unit_tests/test_runner.py @@ -80,16 +80,16 @@ def test_run_ingest_extract_using_all_write_modes( run_ingest(elt_job) call_args_list = mock_iceberg_io.write_table.call_args_list - assert len(call_args_list) == 4 expected_table_names = [ "table_default_write", "table_replace_mode", "table_merge_mode", - "empty", ] - expected_write_modes = ("append", "replace", "merge", "append") - expected_merge_on = ([], [], ["name"], []) + assert len(call_args_list) == len(expected_table_names) + + expected_write_modes = ("append", "replace", "merge") + expected_merge_on = ([], [], ["name"]) for index, call in enumerate(call_args_list): call_args, call_kwargs = call.args, call.kwargs @@ -97,10 +97,7 @@ def test_run_ingest_extract_using_all_write_modes( data = call_args[1] assert isinstance(data, pa.Table) expected_table_name = expected_table_names[index] - if expected_table_name != "empty": - assert data["name"][0].as_py() == expected_table_name - else: - assert data.num_rows == 0 + assert data["name"][0].as_py() == expected_table_name assert call_args[2] == expected_write_modes[index] assert call_kwargs["merge_on"] == expected_merge_on[index]