yau-plant-assistant/api/tools/shiftlog.py

"""The operator shift log: the third evidence source.

Not the historian and not a controlled document - db/008_shift_log.sql says why
at length. What matters here is the consequence: entries are ROWS, never
citations. A citation is a controlled document with a confirmed revision, and
an operator's note is not one however true it is.

WHY THIS IS PLAIN SQL AND NOT CUBE
Cube is the only path to the HISTORIAN, and for good reasons that all concern
imh being a live system we hold read-only on. This table is ours, it is small,
and it is not modelled in Cube. Same pattern as tools/equipment.py and
tools/retrieval.py, which query pg-ai directly.

TIMEZONE: storage is UTC and the conversion to SITE_TIMEZONE happens once, in
the query below. That is the same rule Cube follows for the historian - convert
exactly once, on the way out, and never in a prompt.
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo

import psycopg
from psycopg.rows import dict_row

from config import settings

log = logging.getLogger("tools.shiftlog")


# The shift log is asked for by name. This is a lexical trigger rather than a
# classifier label on purpose: adding a sixth class to a five-way classifier -
# the most safety-relevant component in the stack - to reach a demo table would
# be a poor trade. The cost is that this only fires when the operator uses one
# of these words, which is exactly what the eval case pins.
_MENTIONS = re.compile(
    r"\b(shift\s*log|shift\s*notes?|operator\s*log|log\s*book|logbook|handover)\b",
    re.IGNORECASE,
)


@dataclass
class ShiftLogResult:
    rows: list[dict[str, Any]]
    row_count: int
    time_window: dict[str, str]
    # TRUE when any entry returned is demo fiction. Carried out to the answer
    # as used_fixture_data, for the same reason the historian fixtures do it:
    # an entry nobody wrote must not read like one somebody did.
    used_demo_data: bool


def mentions_shift_log(question: str) -> bool:
    return _MENTIONS.search(question) is not None


def _connect() -> psycopg.Connection:
    cfg = settings()
    return psycopg.connect(
        cfg.dsn(),
        row_factory=dict_row,
        application_name="ai-api",
        connect_timeout=5,
    )


def entries(
    *, days: int, equipment_id: str | None = None, conn: psycopg.Connection | None = None
) -> ShiftLogResult:
    """Shift log entries over a rolling N x 24 h window, oldest first.

    The window matches the historical lane's window rather than anything about
    this table - nothing here expires. When the query window becomes variable
    (REQUESTS.md), this source can reach further back than the historian can.
    """
    cfg = settings()
    tz = ZoneInfo(cfg.site_timezone)
    end = datetime.now(timezone.utc)
    start = end - timedelta(days=days)
    local_end = end.astimezone(tz)

    owned = conn is None
    conn = conn or _connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT to_char(entry_time AT TIME ZONE %(tz)s,
                               'YYYY-MM-DD HH24:MI')       AS entry_time_local,
                       shift, author, equipment_id, entry_text, is_demo
                  FROM shift_log
                 WHERE entry_time >= %(start)s
                   AND entry_time <= %(end)s
                   -- Cast is required: an untyped NULL parameter used on both
                   -- sides of an OR leaves Postgres unable to infer the type.
                   AND (%(equipment_id)s::text IS NULL
                        OR equipment_id = %(equipment_id)s::text)
                 ORDER BY entry_time
                """,
                {"tz": cfg.site_timezone, "start": start, "end": end,
                 "equipment_id": equipment_id},
            )
            rows = [dict(r) for r in cur.fetchall()]
    finally:
        if owned:
            conn.close()

    log.info("shift_log: %d entries over %d days (equipment=%s)",
             len(rows), days, equipment_id or "any")

    return ShiftLogResult(
        rows=rows,
        row_count=len(rows),
        time_window={
            "start": start.astimezone(tz).strftime("%Y-%m-%dT%H:%M:%S"),
            "end": local_end.strftime("%Y-%m-%dT%H:%M:%S"),
            "timezone": cfg.site_timezone,
            "description": (
                f"rolling {days} days to "
                f"{local_end.strftime('%Y-%m-%d %H:%M')} {local_end.tzname()}"
            ),
        },
        used_demo_data=any(r["is_demo"] for r in rows),
    )