|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from decimal import Decimal, InvalidOperation |
| 4 | +from io import BytesIO |
| 5 | +import posixpath |
| 6 | +import re |
| 7 | +from typing import Any |
| 8 | +from xml.etree import ElementTree as ET |
| 9 | +from zipfile import ZipFile |
| 10 | + |
| 11 | + |
| 12 | +_CELL_RE = re.compile(r"([A-Z]+)(\d+)") |
| 13 | + |
| 14 | + |
| 15 | +def parse_pscr_cents_from_xlsx(xlsx_bytes: bytes) -> Decimal: |
| 16 | + """Extract the Rider 18 PSCR value as cents/kWh.""" |
| 17 | + with ZipFile(BytesIO(xlsx_bytes)) as workbook: |
| 18 | + shared_strings = _shared_strings(workbook) |
| 19 | + sheet_path = _rates_sheet_path(workbook) |
| 20 | + rows = _worksheet_rows(workbook, sheet_path, shared_strings) |
| 21 | + |
| 22 | + for row_number in sorted(rows): |
| 23 | + row = rows[row_number] |
| 24 | + label = " ".join(row.values()).lower() |
| 25 | + if "pscr" not in label: |
| 26 | + continue |
| 27 | + |
| 28 | + for column in ("C", "D", "B", "E"): |
| 29 | + cents = _dollars_to_cents(row.get(column)) |
| 30 | + if cents is not None: |
| 31 | + return cents |
| 32 | + |
| 33 | + raise ValueError("Rider 18 workbook does not contain a PSCR value") |
| 34 | + |
| 35 | + |
| 36 | +def _shared_strings(workbook: ZipFile) -> list[str]: |
| 37 | + try: |
| 38 | + root = ET.fromstring(workbook.read("xl/sharedStrings.xml")) |
| 39 | + except KeyError: |
| 40 | + return [] |
| 41 | + |
| 42 | + return ["".join(text.text or "" for text in item.findall(".//{*}t")) for item in root.findall("{*}si")] |
| 43 | + |
| 44 | + |
| 45 | +def _rates_sheet_path(workbook: ZipFile) -> str: |
| 46 | + workbook_root = ET.fromstring(workbook.read("xl/workbook.xml")) |
| 47 | + rels_root = ET.fromstring(workbook.read("xl/_rels/workbook.xml.rels")) |
| 48 | + rel_targets = { |
| 49 | + rel.attrib["Id"]: _normalize_target(rel.attrib.get("Target", "")) |
| 50 | + for rel in rels_root.findall("{*}Relationship") |
| 51 | + if "Id" in rel.attrib |
| 52 | + } |
| 53 | + |
| 54 | + fallback: str | None = None |
| 55 | + for sheet in workbook_root.findall(".//{*}sheet"): |
| 56 | + name = sheet.attrib.get("name", "") |
| 57 | + rel_id = sheet.attrib.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id") |
| 58 | + target = rel_targets.get(rel_id or "") |
| 59 | + if target is None: |
| 60 | + continue |
| 61 | + if name.strip().lower() == "rates and credits": |
| 62 | + return target |
| 63 | + if fallback is None and "rate" in name.lower(): |
| 64 | + fallback = target |
| 65 | + |
| 66 | + if fallback is None: |
| 67 | + raise ValueError("Rider 18 workbook does not contain a rates sheet") |
| 68 | + return fallback |
| 69 | + |
| 70 | + |
| 71 | +def _normalize_target(target: str) -> str: |
| 72 | + if target.startswith("/"): |
| 73 | + return target.lstrip("/") |
| 74 | + return posixpath.normpath(posixpath.join("xl", target)) |
| 75 | + |
| 76 | + |
| 77 | +def _worksheet_rows( |
| 78 | + workbook: ZipFile, |
| 79 | + sheet_path: str, |
| 80 | + shared_strings: list[str], |
| 81 | +) -> dict[int, dict[str, str]]: |
| 82 | + root = ET.fromstring(workbook.read(sheet_path)) |
| 83 | + rows: dict[int, dict[str, str]] = {} |
| 84 | + |
| 85 | + for cell in root.findall(".//{*}c"): |
| 86 | + ref = cell.attrib.get("r", "") |
| 87 | + match = _CELL_RE.match(ref) |
| 88 | + if not match: |
| 89 | + continue |
| 90 | + |
| 91 | + column, row_number_raw = match.groups() |
| 92 | + value = _cell_value(cell, shared_strings) |
| 93 | + if value is not None: |
| 94 | + rows.setdefault(int(row_number_raw), {})[column] = " ".join(value.split()) |
| 95 | + |
| 96 | + return rows |
| 97 | + |
| 98 | + |
| 99 | +def _cell_value(cell: ET.Element, shared_strings: list[str]) -> str | None: |
| 100 | + cell_type = cell.attrib.get("t") |
| 101 | + if cell_type == "inlineStr": |
| 102 | + return "".join(text.text or "" for text in cell.findall(".//{*}t")) |
| 103 | + |
| 104 | + value = cell.find("{*}v") |
| 105 | + if value is None or value.text is None: |
| 106 | + return None |
| 107 | + |
| 108 | + if cell_type == "s": |
| 109 | + try: |
| 110 | + return shared_strings[int(value.text)] |
| 111 | + except (IndexError, ValueError): |
| 112 | + return None |
| 113 | + return value.text |
| 114 | + |
| 115 | + |
| 116 | +def _dollars_to_cents(value: Any) -> Decimal | None: |
| 117 | + text = str(value or "").replace("$", "").replace(",", "").strip() |
| 118 | + if not text: |
| 119 | + return None |
| 120 | + |
| 121 | + try: |
| 122 | + return (abs(Decimal(text)) * Decimal("100")).quantize(Decimal("0.001")) |
| 123 | + except InvalidOperation: |
| 124 | + return None |
0 commit comments