"""Load Sukawarna / Kandang 2 preview sheets into ORM-ready row dicts.""" from __future__ import annotations from datetime import date, datetime from pathlib import Path from typing import Any from openpyxl import load_workbook REPO_ROOT = Path(__file__).resolve().parents[3] PREVIEW_XLSX = REPO_ROOT / "backup" / "kandang5_sukawarna_preview.xlsx" IOT_EXTRACT_JSON = REPO_ROOT / "scripts" / "iot_panel_cycle_extract.json" # Final-cycle report (reference for manual feed use on Kandang 2). FEED_REFERENCE_XLSX = ( REPO_ROOT / "backup" / "Laporan_Akhir-Siklus_Sukawarna_Kandang-Atas_Akhir-Siklus_2026-05-22_sd_2026-07-09.xlsx" ) def as_date(value: Any) -> date | None: if value is None: return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = str(value).strip() if len(text) >= 10 and text[0].isdigit() and text[4] == "-": return date.fromisoformat(text[:10]) return None def as_number(value: Any, default: float | int = 0): if value is None or value == "": return default if isinstance(value, (int, float)) and not isinstance(value, bool): return value try: if isinstance(value, str) and "." not in value: return int(value) return float(value) except (TypeError, ValueError): return default def dated_rows(ws) -> list[dict[str, Any]]: headers = [cell.value for cell in next(ws.iter_rows(min_row=1, max_row=1))] rows: list[dict[str, Any]] = [] for raw in ws.iter_rows(min_row=2, values_only=True): mapped = {str(headers[i]): raw[i] for i in range(len(headers))} parsed = as_date(mapped.get("date")) if parsed is None: continue mapped["date"] = parsed rows.append(mapped) return rows def load_preview(path: Path | None = None) -> dict[str, list[dict[str, Any]]]: workbook = load_workbook(path or PREVIEW_XLSX, data_only=True) return { "chicken_counting": dated_rows(workbook["chicken_counting"]), "chicken_weight": dated_rows(workbook["chicken_weight"]), "feed_sacks": dated_rows(workbook["feed_sacks"]), "manual_input": dated_rows(workbook["manual_input"]), "kpi_preview": dated_rows(workbook["kpi_preview"]), } def load_feed_reference(path: Path | None = None) -> dict[str, dict[str, int]]: """Load the final-cycle report's Feed In / Feed Use sheets. Returns ``{date: {manual_in, iot_in, manual_use, iot_use}}``. These values are the authoritative reference for the karung "Manual" / "IOT" columns. """ workbook = load_workbook(path or FEED_REFERENCE_XLSX, data_only=True) feed_in: dict[str, dict[str, int]] = {} for raw in workbook["Feed In"].iter_rows(min_row=3, values_only=True): parsed = as_date(raw[0]) if parsed is None: continue feed_in[parsed.isoformat()] = { "manual_in": int(as_number(raw[1])), "iot_in": int(as_number(raw[2])), } feed_use: dict[str, dict[str, int]] = {} for raw in workbook["Feed Use"].iter_rows(min_row=3, values_only=True): parsed = as_date(raw[0]) if parsed is None: continue feed_use[parsed.isoformat()] = { "manual_use": int(as_number(raw[2])), "iot_use": int(as_number(raw[3])), } merged: dict[str, dict[str, int]] = {} for key in feed_in: if key in feed_use: merged[key] = {**feed_in[key], **feed_use[key]} return merged