feat(elt-pipelines): Port electricity_sharepoint pipeline to elt

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.

elt-common/tests/unit_tests/test_runner.py

@@ -80,27 +80,24 @@ 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
         assert call_args[0] == _test_elt_job_table_id(elt_job, expected_table_names[index])
         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]

elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/electricity_sharepoint.py

@@ -0,0 +1,141 @@
+"""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
+
+import pandas as pd
+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
+
+SITE_URL = "https://stfc365.sharepoint.com/sites/ISISSustainability"
+MAX_WORKERS = min(8, (os.cpu_count() or 1) + 4)
+
+LOGGER = logging.getLogger(__name__)
+
+_root_path = "/General/RDM Data"
+_default_backfill_globs = [
+    "**/*.xlsx",
+    "**/*-daily.csv",
+    "**/*-manual-export.csv",
+]
+
+
+class Configuration(M365Credentials):
+    backfill: bool = False
+    backfill_globs: list[str] = []
+
+    @property
+    def glob_patterns(self):
+        if not self.backfill:
+            return ["*-ISIS.csv"]
+        return self.backfill_globs if self.backfill_globs else _default_backfill_globs
+
+
+class Extract(BaseExtract):
+    config_cls = Configuration
+
+    def __init__(self, cfg: Configuration):
+        super().__init__(cfg)
+        self._client = SPListClient(SITE_URL, cfg)
+        self._backfilling = cfg.backfill
+        self._glob_patterns = cfg.glob_patterns
+
+        LOGGER.debug(f"Searching for files matching: {self._glob_patterns}")
+
+    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",
+            ),
+        )
+
+    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"
+                )
+            else:
+                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
+                )
+            )
+
+        LOGGER.debug(f"Matched {len(files)} files")
+
+        resolved = self._read_files(files)
+        df = read_contents_to_dataframe(resolved)
+        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 = []
+        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
+
+
+def read_contents_to_dataframe(files: list[tuple[str, bytes]]):
+    """Extracts data from files into a single, combined dataframe
+
+    :param files: (name, content) pairs for the files to extract data from
+    """
+
+    def read_as_dataframe(file_name, file_bytes) -> pd.DataFrame | None:
+        file_content = io.BytesIO(file_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_name, file_content)
+            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: 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))
+
+    return df_batch

elt-pipelines/facility_ops/ingest/estates/electricity_sharepoint/parsing.py

@@ -0,0 +1,150 @@
+"""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]
+    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":
+            # 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")

elt-pipelines/pyproject.toml

@@ -15,7 +15,7 @@ statusdisplay = [
     "requests>=2.34.2",
 ]
 
-accelerator_sharepoint = [
+sharepoint = [
     "pandas >= 3.0.3",
     "openpyxl >= 3.1.5",
     "elt-common[m365]"