add water history data

src/ems_water_grp/backfill_water_pg_1day.py

+# -*- coding:utf-8 -*-
+"""
+Backfill water/pressure daily history points from Influx origin queryPeriod.
+
+Usage:
+    python -m ems_water_grp.backfill_water_pg_1day \
+        --start "2025-07-13 00:00:00" \
+        --end "2025-07-13 23:59:59"
+
+Notes:
+    - Water meter:
+        water_vol = last meter value of the day - first meter value of the day
+        meter_value = last meter value of the day
+    - Pressure gauge:
+        pg_vol = average value of all points in the day
+        meter_value = last value of the day
+"""
+
+import argparse
+import asyncio
+import json
+from collections import defaultdict
+from decimal import Decimal, ROUND_HALF_UP
+from typing import Dict, Iterable, List, Optional, Tuple
+
+import pendulum
+
+from ems_water_grp.constants import WATER_PG_MAP, WATER_SCADA_MAP
+from infra.config.settings import SETTING
+from infra.http.httpx_util import HttpxClient
+from infra.logger.logger import Logger
+from infra.mysql.aiomysql_util import MysqlUtil
+from utils.time_format import CST, YMD_Hms
+
+
+log_name = "backfill_water_pg_1day"
+Logger.init_logger_path("./ems_water_grp", f"{log_name}.log", log_name)
+logger = Logger.getLogger(log_name)
+
+
+INFLUX_PERIOD_URL = "http://172.18.4.77/influx/origin/queryPeriod"
+
+WATER_TABLE = "water_1day_point"
+PG_TABLE = "pg_1day_point"
+
+
+def parse_dt(value: str) -> pendulum.DateTime:
+    return pendulum.parse(value, tz=CST)
+
+
+def fmt_dt(value: pendulum.DateTime) -> str:
+    return value.in_timezone(CST).format(YMD_Hms)
+
+
+def iter_days(
+        start: pendulum.DateTime,
+        end: pendulum.DateTime,
+) -> Iterable[Tuple[pendulum.DateTime, pendulum.DateTime, pendulum.DateTime]]:
+    cur = start.start_of("day")
+    end_day = end.start_of("day")
+    while cur <= end_day:
+        day_start = start if cur < start.start_of("day") else cur
+        if cur == start.start_of("day") and start > cur:
+            day_start = start
+
+        cur_end = cur.end_of("day")
+        day_end = end if cur == end_day and end < cur_end else cur_end
+        yield cur, day_start, day_end
+        cur = cur.add(days=1)
+
+
+def build_fac_ids() -> List[dict]:
+    fac_map = defaultdict(list)
+    for node_id, meta in WATER_SCADA_MAP.items():
+        fac_map[meta["cid"]].append(node_id)
+
+    return [
+        {"factoryId": factory_id, "ids": ids}
+        for factory_id, ids in sorted(fac_map.items())
+    ]
+
+
+def build_period_payload(
+        day_start: pendulum.DateTime,
+        day_end: pendulum.DateTime,
+) -> dict:
+    return {
+        "facIds": build_fac_ids(),
+        "sourceDtStart": fmt_dt(day_start),
+        "sourceDtEnd": fmt_dt(day_end),
+        "sorting": "acs",
+        "IsFillNullForMin": 0,
+        "dataTypeId": 1,
+        "isPostProcess": 0,
+        "isDynamicAdjust": 0,
+    }
+
+
+def to_decimal(value) -> Optional[Decimal]:
+    try:
+        if value is None or value == "":
+            return None
+        return Decimal(str(value))
+    except Exception:
+        return None
+
+
+def quantize_3(value: Decimal) -> Decimal:
+    return value.quantize(Decimal("0.001"), rounding=ROUND_HALF_UP)
+
+
+def normalize_items(items: List[dict]) -> List[Tuple[pendulum.DateTime, Decimal]]:
+    points = []
+    for item in items:
+        value = to_decimal(item.get("value"))
+        if value is None:
+            continue
+
+        raw_time = item.get("time")
+        if not raw_time:
+            continue
+
+        try:
+            point_time = pendulum.parse(raw_time, tz=CST)
+        except Exception:
+            logger.warning(f"Skip invalid time item: {item}")
+            continue
+
+        points.append((point_time, value))
+
+    points.sort(key=lambda x: x[0])
+    return points
+
+
+class DeviceResolver:
+    """
+    Resolve Influx node id -> monitor id -> water/pg master id.
+
+    The project only keeps node_id -> sid/cid in constants.py.  The daily
+    tables require numeric water_id/pg_id and mtid, so they must be loaded
+    from MySQL.
+    """
+
+    def __init__(
+            self,
+            db: str,
+            monitor_table: str,
+            water_master_table: str,
+            pg_master_table: str,
+    ):
+        self.db = db
+        self.monitor_table = monitor_table
+        self.water_master_table = water_master_table
+        self.pg_master_table = pg_master_table
+        self._column_cache = {}
+
+    async def has_column(self, conn: MysqlUtil, table: str, column: str) -> bool:
+        key = (table, column)
+        if key in self._column_cache:
+            return self._column_cache[key]
+
+        sql = """
+            SELECT COUNT(*) AS cnt
+            FROM information_schema.columns
+            WHERE table_schema = %s
+              AND table_name = %s
+              AND column_name = %s
+        """
+        row = await conn.fetchone(sql, (self.db, table, column))
+        exists = bool(row and row.get("cnt"))
+        self._column_cache[key] = exists
+        return exists
+
+    async def load_mtid_by_sid(self, conn: MysqlUtil, sid: str) -> Optional[int]:
+        sql = f"SELECT mtid FROM `{self.monitor_table}` WHERE sid = %s LIMIT 1"
+        row = await conn.fetchone(sql, (sid,))
+        if not row:
+            return None
+        return row.get("mtid")
+
+    async def load_device_id(
+            self,
+            conn: MysqlUtil,
+            table: str,
+            mtid: int,
+            sid: str,
+            id_column: str,
+    ) -> Optional[int]:
+        if await self.has_column(conn, table, "mtid"):
+            row = await conn.fetchone(
+                f"SELECT `{id_column}` FROM `{table}` WHERE mtid = %s LIMIT 1",
+                (mtid,),
+            )
+            if row:
+                return row.get(id_column)
+
+        if await self.has_column(conn, table, "sid"):
+            row = await conn.fetchone(
+                f"SELECT `{id_column}` FROM `{table}` WHERE sid = %s LIMIT 1",
+                (sid,),
+            )
+            if row:
+                return row.get(id_column)
+
+        return None
+
+    async def resolve_all(self) -> Dict[str, dict]:
+        resolved = {}
+        async with MysqlUtil(self.db) as conn:
+            for node_id, meta in WATER_SCADA_MAP.items():
+                sid = meta["sid"]
+                cid = meta["cid"]
+                is_pg = node_id in WATER_PG_MAP
+                master_table = (
+                    self.pg_master_table if is_pg else self.water_master_table
+                )
+                id_column = "pg_id" if is_pg else "water_id"
+
+                mtid = await self.load_mtid_by_sid(conn, sid)
+                if not mtid:
+                    logger.error(f"Missing monitor by sid={sid}, node_id={node_id}")
+                    continue
+
+                device_id = await self.load_device_id(
+                    conn, master_table, mtid, sid, id_column
+                )
+                if not device_id:
+                    logger.error(
+                        f"Missing {id_column} in {master_table}, "
+                        f"sid={sid}, mtid={mtid}, node_id={node_id}"
+                    )
+                    continue
+
+                resolved[node_id] = {
+                    "sid": sid,
+                    "cid": cid,
+                    "mtid": mtid,
+                    "device_id": device_id,
+                    "is_pg": is_pg,
+                }
+
+        logger.info(
+            f"Resolved {len(resolved)}/{len(WATER_SCADA_MAP)} water/pg devices"
+        )
+        return resolved
+
+
+class WaterPgDailyBackfill:
+    def __init__(
+            self,
+            db: str,
+            monitor_table: str,
+            water_master_table: str,
+            pg_master_table: str,
+            dry_run: bool,
+    ):
+        self.db = db
+        self.dry_run = dry_run
+        self.http = HttpxClient()
+        self.resolver = DeviceResolver(
+            db=db,
+            monitor_table=monitor_table,
+            water_master_table=water_master_table,
+            pg_master_table=pg_master_table,
+        )
+
+    async def close(self):
+        await self.http.close()
+        await MysqlUtil.close_all_pools()
+
+    async def fetch_period(
+            self,
+            day_start: pendulum.DateTime,
+            day_end: pendulum.DateTime,
+    ) -> List[dict]:
+        payload = build_period_payload(day_start, day_end)
+        resp_str, status_code = await self.http.post(
+            INFLUX_PERIOD_URL,
+            headers={"Content-Type": "application/json"},
+            json=payload,
+        )
+        if status_code != 200:
+            raise RuntimeError(f"Influx fetch failed: HTTP {status_code}, {resp_str}")
+
+        resp = json.loads(resp_str)
+        if resp.get("code") != 0:
+            raise RuntimeError(f"Influx fetch failed: {resp}")
+
+        return resp.get("data") or []
+
+    def calc_day_rows(
+            self,
+            create_time: pendulum.DateTime,
+            data: List[dict],
+            devices: Dict[str, dict],
+    ) -> Tuple[List[tuple], List[tuple]]:
+        water_rows = []
+        pg_rows = []
+
+        for node in data:
+            node_id = node.get("id")
+            device = devices.get(node_id)
+            if not device:
+                continue
+
+            points = normalize_items(node.get("items") or [])
+            if not points:
+                logger.warning(f"No valid points for node_id={node_id}")
+                continue
+
+            first_time, first_value = points[0]
+            last_time, last_value = points[-1]
+            create_time_str = fmt_dt(create_time.start_of("day"))
+            read_time_str = fmt_dt(last_time)
+
+            if device["is_pg"]:
+                avg_value = (
+                    sum((value for _, value in points), Decimal("0"))
+                    / Decimal(len(points))
+                )
+                pg_rows.append((
+                    device["device_id"],
+                    device["mtid"],
+                    device["cid"],
+                    create_time_str,
+                    quantize_3(avg_value),
+                    quantize_3(last_value),
+                    read_time_str,
+                ))
+            else:
+                water_vol = last_value - first_value
+                water_rows.append((
+                    device["device_id"],
+                    device["mtid"],
+                    device["cid"],
+                    create_time_str,
+                    float(water_vol),
+                    float(last_value),
+                    read_time_str,
+                ))
+
+        return water_rows, pg_rows
+
+    async def save_rows(self, water_rows: List[tuple], pg_rows: List[tuple]):
+        if self.dry_run:
+            logger.info(
+                f"Dry run, skip insert. water_rows={len(water_rows)}, "
+                f"pg_rows={len(pg_rows)}"
+            )
+            return
+
+        water_sql = f"""
+            INSERT INTO `{WATER_TABLE}`
+                (water_id, mtid, cid, create_time, water_vol, meter_value, read_time)
+            VALUES (%s, %s, %s, %s, %s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                mtid = VALUES(mtid),
+                cid = VALUES(cid),
+                water_vol = VALUES(water_vol),
+                meter_value = VALUES(meter_value),
+                read_time = VALUES(read_time)
+        """
+        pg_sql = f"""
+            INSERT INTO `{PG_TABLE}`
+                (pg_id, mtid, cid, create_time, pg_vol, meter_value, read_time)
+            VALUES (%s, %s, %s, %s, %s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                mtid = VALUES(mtid),
+                cid = VALUES(cid),
+                pg_vol = VALUES(pg_vol),
+                meter_value = VALUES(meter_value),
+                read_time = VALUES(read_time)
+        """
+
+        async with MysqlUtil(self.db) as conn:
+            if water_rows:
+                await conn.insert_many(water_sql, water_rows)
+            if pg_rows:
+                await conn.insert_many(pg_sql, pg_rows)
+
+    async def run(self, start: pendulum.DateTime, end: pendulum.DateTime):
+        if end < start:
+            raise ValueError("end must be greater than or equal to start")
+
+        devices = await self.resolver.resolve_all()
+        if not devices:
+            raise RuntimeError("No water/pg devices resolved from MySQL")
+
+        total_water = 0
+        total_pg = 0
+        for create_time, day_start, day_end in iter_days(start, end):
+            logger.info(f"Fetch period {fmt_dt(day_start)} ~ {fmt_dt(day_end)}")
+            data = await self.fetch_period(day_start, day_end)
+            water_rows, pg_rows = self.calc_day_rows(create_time, data, devices)
+            await self.save_rows(water_rows, pg_rows)
+
+            total_water += len(water_rows)
+            total_pg += len(pg_rows)
+            logger.info(
+                f"Saved day={create_time.to_date_string()}, "
+                f"water={len(water_rows)}, pg={len(pg_rows)}"
+            )
+
+        logger.info(f"Backfill finished. total_water={total_water}, total_pg={total_pg}")
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Backfill water_1day_point and pg_1day_point from InfluxDB."
+    )
+    parser.add_argument("--start", required=True, help="Start time, e.g. 2025-07-01 00:00:00")
+    parser.add_argument("--end", required=True, help="End time, e.g. 2025-07-31 23:59:59")
+    parser.add_argument("--db", default="water_works", help="MySQL database name")
+    parser.add_argument("--monitor-table", default="monitor")
+    parser.add_argument("--water-master-table", default="water")
+    parser.add_argument("--pg-master-table", default="water_pg")
+    parser.add_argument("--dry-run", action="store_true", help="Fetch and calculate only")
+    return parser
+
+
+async def async_main():
+    args = build_arg_parser().parse_args()
+    logger.info(
+        f"MySQL host={SETTING.mysql_host}, db={args.db}, "
+        f"start={args.start}, end={args.end}, dry_run={args.dry_run}"
+    )
+
+    service = WaterPgDailyBackfill(
+        db=args.db,
+        monitor_table=args.monitor_table,
+        water_master_table=args.water_master_table,
+        pg_master_table=args.pg_master_table,
+        dry_run=args.dry_run,
+    )
+    try:
+        await service.run(parse_dt(args.start), parse_dt(args.end))
+    finally:
+        await service.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(async_main())