diff --git a/agents/analysis/.DS_Store b/agents/analysis/.DS_Store deleted file mode 100644 index a3a7e4aee966a112a2854f47ff0d6b6134342a86..0000000000000000000000000000000000000000 Binary files a/agents/analysis/.DS_Store and /dev/null differ diff --git a/agents/analysis/.env.example b/agents/analysis/.env.example deleted file mode 100644 index f41cd816fbfa2aca8ece9dc88c553eeac4baf2e3..0000000000000000000000000000000000000000 --- a/agents/analysis/.env.example +++ /dev/null @@ -1,7 +0,0 @@ -GROQ_API_KEY= -GROQ_MODEL=openai/gpt-oss-120b -DATABASE_URL= -LOCAL_TIMEZONE=Asia/Karachi -LOCAL_TIME_OFFSET_HOURS=3 -OPENREMOTE_SCHEMA=openremote -NEXT_PUBLIC_API_BASE_URL=http://localhost:8000 diff --git a/agents/analysis/.gitignore b/agents/analysis/.gitignore deleted file mode 100644 index f8ac107719bbf9733af860932b917d0ba030c97e..0000000000000000000000000000000000000000 --- a/agents/analysis/.gitignore +++ /dev/null @@ -1,9 +0,0 @@ -.venv/ -__pycache__/ -.pytest_cache/ -.next/ -node_modules/ -.env -dist/ -build/ -*.pyc diff --git a/agents/analysis/backend/app/__init__.py b/agents/analysis/backend/app/__init__.py deleted file mode 100644 index d3f5a12faa99758192ecc4ed3fc22c9249232e86..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/__init__.py +++ /dev/null @@ -1 +0,0 @@ - diff --git a/agents/analysis/backend/app/__pycache__/__init__.cpython-313.pyc b/agents/analysis/backend/app/__pycache__/__init__.cpython-313.pyc deleted file mode 100644 index 0c927a686097c38b6718ecb1e181634aff3845a4..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/__pycache__/__init__.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/__pycache__/main.cpython-313.pyc b/agents/analysis/backend/app/__pycache__/main.cpython-313.pyc deleted file mode 100644 index d938345f5831065eba74bcfddd7ff7ccecbaf708..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/__pycache__/main.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/__init__.py b/agents/analysis/backend/app/api/routes/__init__.py deleted file mode 100644 index d3f5a12faa99758192ecc4ed3fc22c9249232e86..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/api/routes/__init__.py +++ /dev/null @@ -1 +0,0 @@ - diff --git a/agents/analysis/backend/app/api/routes/__pycache__/__init__.cpython-313.pyc b/agents/analysis/backend/app/api/routes/__pycache__/__init__.cpython-313.pyc deleted file mode 100644 index e1c7f2f14b5b482856f936383e98c7e5dc00815b..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/api/routes/__pycache__/__init__.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/__pycache__/analysis.cpython-313.pyc b/agents/analysis/backend/app/api/routes/__pycache__/analysis.cpython-313.pyc deleted file mode 100644 index 8f97db6fa3da94af81f87ae88c1665314c671b77..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/api/routes/__pycache__/analysis.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/__pycache__/assets.cpython-313.pyc b/agents/analysis/backend/app/api/routes/__pycache__/assets.cpython-313.pyc deleted file mode 100644 index cba797f0549e32c9cfee2930574d976497e4624f..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/api/routes/__pycache__/assets.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/__pycache__/chat.cpython-313.pyc b/agents/analysis/backend/app/api/routes/__pycache__/chat.cpython-313.pyc deleted file mode 100644 index 8a7945f17c700390b20ce1d3786f9cfd42ffabcf..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/api/routes/__pycache__/chat.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/__pycache__/schedules.cpython-313.pyc b/agents/analysis/backend/app/api/routes/__pycache__/schedules.cpython-313.pyc deleted file mode 100644 index 2bdf5ae0d1f5f3864abae0edfa1c7b2ffdd83c77..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/api/routes/__pycache__/schedules.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/api/routes/analysis.py b/agents/analysis/backend/app/api/routes/analysis.py deleted file mode 100644 index 5211dc99f1a92f9476f3a9139dbe0c93e67dbe0e..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/api/routes/analysis.py +++ /dev/null @@ -1,12 +0,0 @@ -from fastapi import APIRouter - -from agents.analysis.backend.app.models.schemas import AnalysisRequest, AnalysisResponse -from agents.analysis.backend.app.services.analysis_agent import analysis_agent_service - - -router = APIRouter(tags=["analysis"]) - - -@router.post("/analysis/run", response_model=AnalysisResponse) -async def run_analysis(payload: AnalysisRequest) -> AnalysisResponse: - return await analysis_agent_service.run_analysis(payload) diff --git a/agents/analysis/backend/app/api/routes/assets.py b/agents/analysis/backend/app/api/routes/assets.py deleted file mode 100644 index 5908e13a6b2314cd8fbd72ee0bb715b8f82ad36d..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/api/routes/assets.py +++ /dev/null @@ -1,19 +0,0 @@ -from fastapi import APIRouter, HTTPException - -from agents.analysis.backend.app.services.live_asset_registry import live_asset_registry - - -router = APIRouter(tags=["assets"]) - - -@router.get("/assets") -async def list_assets(user_id: str = "user_1") -> list[dict]: - return [asset.model_dump() for asset in live_asset_registry.list_assets(user_id=user_id)] - - -@router.get("/assets/{asset_id}") -async def get_asset(asset_id: str, user_id: str = "user_1") -> dict: - asset = live_asset_registry.get_asset(asset_id, user_id=user_id) - if asset is None: - raise HTTPException(status_code=404, detail=f"Unknown asset: {asset_id}") - return asset.model_dump() diff --git a/agents/analysis/backend/app/api/routes/chat.py b/agents/analysis/backend/app/api/routes/chat.py deleted file mode 100644 index 2fd62d661ee804608248e6c79428f5ec1c2a1ebb..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/api/routes/chat.py +++ /dev/null @@ -1,12 +0,0 @@ -from fastapi import APIRouter - -from agents.analysis.backend.app.models.schemas import ChatRequest, ChatResponse -from agents.analysis.backend.app.services.analysis_agent import analysis_agent_service - - -router = APIRouter(tags=["chat"]) - - -@router.post("/chat/message", response_model=ChatResponse) -async def chat_message(payload: ChatRequest) -> ChatResponse: - return await analysis_agent_service.run_chat(payload) diff --git a/agents/analysis/backend/app/api/routes/schedules.py b/agents/analysis/backend/app/api/routes/schedules.py deleted file mode 100644 index 96d911e06ded50c373a4b02bc72ee549f59fa5b3..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/api/routes/schedules.py +++ /dev/null @@ -1,42 +0,0 @@ -from fastapi import APIRouter, HTTPException -from typing import List, Optional -from pydantic import BaseModel - -from agents.analysis.backend.app.models.schemas import ChatResponse -from agents.analysis.backend.app.services.scheduling_agent import n8n_fast_path_service - -router = APIRouter(tags=["schedules"]) - -class ScheduleInfo(BaseModel): - id: str - name: str - description: str - triggers: List[str] - -@router.get("/schedules", response_model=List[ScheduleInfo]) -async def list_schedules(): - """List all available scheduled report types.""" - return [ - ScheduleInfo( - id="th_monitoring", - name="T & H Monitoring Report", - description="Generates a comprehensive Temperature and Humidity report for all coldrooms and labs.", - triggers=["Scheduled T&H Monitoring", "T&H Monitoring Schedule"] - ), - ScheduleInfo( - id="filling_machines", - name="Filling Machines Production Report", - description="Generates a production summary for all filling machines across shifts.", - triggers=["Schedule Filling Machines", "Scheduled Filling Machines"] - ) - ] - -@router.post("/schedules/run/{schedule_id}", response_model=ChatResponse) -async def run_schedule(schedule_id: str, offset: int = 0, span_days: int = 1): - """Manually trigger a scheduled report by its ID.""" - if schedule_id == "th_monitoring": - return n8n_fast_path_service._handle_th(offset=offset, span_days=span_days) - elif schedule_id == "filling_machines": - return n8n_fast_path_service._handle_filling(offset=offset, span_days=span_days) - else: - raise HTTPException(status_code=404, detail=f"Schedule '{schedule_id}' not found.") diff --git a/agents/analysis/backend/app/core/__pycache__/config.cpython-313.pyc b/agents/analysis/backend/app/core/__pycache__/config.cpython-313.pyc deleted file mode 100644 index 57dc6e3b8ebe151f961a27f0818632c36530ef69..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/core/__pycache__/config.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/core/config.py b/agents/analysis/backend/app/core/config.py deleted file mode 100644 index 8b8eb9613ed999703aad8acf338013f511643906..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/core/config.py +++ /dev/null @@ -1,28 +0,0 @@ -from pathlib import Path -from functools import lru_cache - -from pydantic_settings import BaseSettings, SettingsConfigDict - - -ROOT_ENV_FILE = Path(__file__).resolve().parents[5] / ".env" - - -class Settings(BaseSettings): - app_name: str = "VisBot Analysis" - groq_api_key: str = "" - groq_model: str = "lgpt-oss-120b" - database_url: str = "sqlite+pysqlite:///:memory:" - local_timezone: str = "Asia/Karachi" - local_time_offset_hours: int = 3 - openremote_schema: str = "openremote" - - model_config = SettingsConfigDict( - env_file=(str(ROOT_ENV_FILE), ".env"), - env_file_encoding="utf-8", - extra="ignore", - ) - - -@lru_cache -def get_settings() -> Settings: - return Settings() diff --git a/agents/analysis/backend/app/main.py b/agents/analysis/backend/app/main.py deleted file mode 100644 index 2bc278da42b0f243805e085e550112756187072a..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/main.py +++ /dev/null @@ -1,33 +0,0 @@ -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware - -from agents.analysis.backend.app.api.routes.analysis import router as analysis_router -from agents.analysis.backend.app.api.routes.assets import router as assets_router -from agents.analysis.backend.app.api.routes.chat import router as chat_router -from agents.analysis.backend.app.api.routes.schedules import router as schedules_router - - -app = FastAPI( - title="VisBot Analysis", - version="0.1.0", - description="Intelligent asset analysis agent powered by FastAPI, LangChain, and Groq.", -) - -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - - -@app.get("/health") -async def health() -> dict[str, str]: - return {"status": "ok"} - - -app.include_router(assets_router, prefix="/api/v1") -app.include_router(chat_router, prefix="/api/v1") -app.include_router(analysis_router, prefix="/api/v1") -app.include_router(schedules_router, prefix="/api/v1") diff --git a/agents/analysis/backend/app/models/__init__.py b/agents/analysis/backend/app/models/__init__.py deleted file mode 100644 index d3f5a12faa99758192ecc4ed3fc22c9249232e86..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/models/__init__.py +++ /dev/null @@ -1 +0,0 @@ - diff --git a/agents/analysis/backend/app/models/__pycache__/__init__.cpython-313.pyc b/agents/analysis/backend/app/models/__pycache__/__init__.cpython-313.pyc deleted file mode 100644 index 276bb621c8e00ae815929f8be6db13f1ede22aad..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/models/__pycache__/__init__.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/models/__pycache__/schemas.cpython-313.pyc b/agents/analysis/backend/app/models/__pycache__/schemas.cpython-313.pyc deleted file mode 100644 index 920cb9946af5a4e8b0919f2f69ccd6aee5468e67..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/models/__pycache__/schemas.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/models/schemas.py b/agents/analysis/backend/app/models/schemas.py deleted file mode 100644 index d4d76840f76e097f0ab31b00d7742d3dd490e72d..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/models/schemas.py +++ /dev/null @@ -1,119 +0,0 @@ -from __future__ import annotations - -from datetime import datetime -from typing import Literal -from uuid import uuid4 - -from pydantic import BaseModel, Field - - -AnalysisScope = Literal["live", "historical"] -WindowUnit = Literal["minutes", "hours", "days", "weeks", "months"] - - -class TimeWindow(BaseModel): - scope: AnalysisScope = "live" - value: int = 1 - unit: WindowUnit = "hours" - label: str = "last 1 hour" - start_at: str | None = None - end_at: str | None = None - - -class AssetProfile(BaseModel): - asset_id: str - name: str - asset_type: str - description: str - db_asset_id: str = "" - parent_name: str = "" - child_count: int = 0 - is_device: bool = False - attribute_keys: list[str] = Field(default_factory=list) - has_live_data: bool = False - data_keys: list[str] = Field(default_factory=list) - supported_analyses: list[str] = Field(default_factory=list) - metrics: list[str] = Field(default_factory=list) - source_table: str = "" - analysis_instructions: str = "" - sql_notes: str = "" - lookup_names: list[str] = Field(default_factory=list) - temperature_keys: list[str] = Field(default_factory=list) - humidity_keys: list[str] = Field(default_factory=list) - level_keys: list[str] = Field(default_factory=list) - production_keys: list[str] = Field(default_factory=list) - nominal_range: dict[str, float] = Field(default_factory=dict) - - -class AnalysisPlan(BaseModel): - analysis_name: str - reasoning: str - time_window: TimeWindow - query_started_at: str = "" - query_ended_at: str = "" - query_window_label: str = "" - planner_prompt: str = "" - sql_prompt: str = "" - analyst_prompt: str = "" - sql_query: str - chart_hints: list[str] = Field(default_factory=list) - warnings: list[str] = Field(default_factory=list) - total_tokens: int = 0 - llm_raw_response: str = "" - - -class AnalysisReport(BaseModel): - report_id: str = Field(default_factory=lambda: str(uuid4())) - title: str - summary: str - markdown: str - html: str - created_at: datetime = Field(default_factory=datetime.utcnow) - - -class AnalysisRequest(BaseModel): - asset_id: str | None = None - question: str - analysis_name: str | None = None - time_window: TimeWindow | None = None - - -class TrendPoint(BaseModel): - label: str - value: float - - -class TrendSeries(BaseModel): - key: str - label: str - color: str - points: list[TrendPoint] = Field(default_factory=list) - - -class TrendChart(BaseModel): - title: str - chart_type: Literal["line", "bar"] - x_label: str - y_label: str - summary: str = "" - series: list[TrendSeries] = Field(default_factory=list) - - -class AnalysisResponse(BaseModel): - asset: AssetProfile - plan: AnalysisPlan - report: AnalysisReport - trend_chart: TrendChart | None = None - trend_charts: list[TrendChart] = Field(default_factory=list) - rows: list[dict] = Field(default_factory=list) - - -class ChatRequest(BaseModel): - message: str - user_id: str = "user_1" - asset_id: str | None = None - - -class ChatResponse(BaseModel): - answer: str - analysis: AnalysisResponse diff --git a/agents/analysis/backend/app/prompts/analysis_planner.txt b/agents/analysis/backend/app/prompts/analysis_planner.txt deleted file mode 100644 index 70accca5727e93a8d883e9bd20c9f2b865722a89..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/prompts/analysis_planner.txt +++ /dev/null @@ -1,24 +0,0 @@ -You are VisBot Analysis, an industrial asset analysis planner. - -Your job: -- understand the user's asset analysis request -- use the asset selected from the user's chat message -- use the backend-provided asset analysis instructions -- select the best analysis type -- infer the time window -- produce a safe SQL query plan - -Rules: -- prefer deterministic analysis types from the asset profile -- never invent unavailable metrics -- keep SQL read-only -- use the asset description to choose the correct analysis style -- explain the result in simple user-friendly language -- if the user asks for live data, prioritize the latest records -- if the user asks for historical analysis, use the requested time window - -Return concise planning text that explains: -1. chosen analysis -2. why it matches the asset -3. what time window should be used -4. what metrics matter most diff --git a/agents/analysis/backend/app/prompts/request_planner.txt b/agents/analysis/backend/app/prompts/request_planner.txt deleted file mode 100644 index d91b41b412e5832d2a1d721cea2e45c1acbd3b98..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/prompts/request_planner.txt +++ /dev/null @@ -1,37 +0,0 @@ -You are VisBot Analysis. - -Your job is to read the user's message and produce a structured asset-analysis plan. - -You must: -- detect the exact asset from the provided asset catalog -- choose the most suitable analysis type -- infer the time window from the user's wording -- prefer historical analysis styles when the user asks for trends, patterns, previous windows, comparisons, changes, drops, rises, stability, or insights over time -- keep the output strictly in JSON - -Rules: -- do not invent assets that are not in the catalog -- the provided asset catalog is a compact shortlist, so use the strongest name, alias, parent, and telemetry-key clues available -- prefer exact asset name matches while ignoring spaces and case -- pay attention to parent_name, child_count, has_live_data, and data_keys -- if the matched item looks like a parent/group asset without direct data, you may still choose it if the user is asking about that system or area -- use the asset description and analysis instructions to decide the analysis style -- if the asset is unclear, return asset_id as an empty string and explain the issue briefly -- if the user asks for "latest", treat it as a live or recent window -- if the user asks for history, trend, previous hour/day/week/month, comparison, movement, or insight, choose a historical window and an analysis type that benefits from multiple returned rows -- if the user asks for analysis but does not mention a time, date, range, or duration, use a historical window of the last 3 hours from the request time -- for cold rooms under T & H Monitoring, prefer `consumption_analysis` for the requested time window even if the user simply asks for analysis or consumption -- keep labels human-readable - -Return JSON with this exact shape: -{{ - "asset_id": "string", - "analysis_name": "string", - "time_window": {{ - "scope": "live or historical", - "value": 1, - "unit": "minutes|hours|days|weeks|months", - "label": "string" - }}, - "reasoning": "short string" -}} diff --git a/agents/analysis/backend/app/prompts/result_analyst.txt b/agents/analysis/backend/app/prompts/result_analyst.txt deleted file mode 100644 index d9d922d65c3b86fa32b6e0b21d279995f4ea2846..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/prompts/result_analyst.txt +++ /dev/null @@ -1,34 +0,0 @@ -You are VisBot Analysis. - -You are given: -- the user's original request -- the matched asset and its analysis instructions -- the SQL query that fetched data -- deterministic findings -- the returned database rows - -Write a clean end-user analysis in simple language. - -Rules: -- be analytical, not generic -- use the actual returned data -- highlight the most important findings first -- mention problems clearly if thresholds are crossed or data is stale -- avoid raw JSON formatting in the main answer -- keep the answer well structured and readable -- keep the answer concise and precise (prefer 5-10 short lines total) -- do not output empty bullet markers, standalone `*`, or markdown separators like `---` -- prefer short headings, clean paragraphs, and real lists only when each bullet has content -- if the selected asset is a parent/group with multiple child assets, summarize all child assets (not just one) and call out only the most critical outliers -- when the request is historical, explicitly describe trend direction, range, highs, lows, movement, and stability using the returned rows -- provide real insights from the historical series instead of restating a single latest point -- for tanks, present the reported metric as percentage filled (e.g. 85%) whenever possible -- for tanks, use `percentage_filled = ((MaxFt - TankOilLevelInFeet001) / MaxFt) * 100` whenever MaxFt is available -- for tanks, do not explain or mention how you calculated the level or percentage, just report the final fill percentage and current depth -- if deterministic findings confirm that MaxFt was found, prioritize the percentage filled in your summary -- for temperature and humidity (T&H) monitoring, always report how much time the system was on or consuming power within the "6am-6am" day -- for all day-based calculations, a 24-hour day is strictly considered from 6:00 AM to 6:00 AM the following day -- for temperature monitoring, when the temperature starts to increase and goes above 20, consider the subsequent temperature dip (after it started to rise) as the indicator that the system turned off at that time -- for temperature compliance, use only thresholds explicitly provided in the asset context or deterministic findings -- do not use generic or assumed industry ranges (for example "usually 2-8C") unless the user explicitly asks for benchmark ranges -- finish with one useful suggested next step diff --git a/agents/analysis/backend/app/prompts/sql_generator.txt b/agents/analysis/backend/app/prompts/sql_generator.txt deleted file mode 100644 index 093c6709063139f2768ee096734e75cf1ff24c87..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/prompts/sql_generator.txt +++ /dev/null @@ -1,46 +0,0 @@ -You are VisBot Analysis. - -Generate exactly one read-only PostgreSQL query for OpenRemote. - -Database rules: -- all tables are under the openremote schema -- the main telemetry tables are openremote.asset and openremote.asset_datapoint -- join them using `ad.entity_id = a.id` -- sensor telemetry is usually stored in asset_datapoint.value JSON -- many telemetry rows use `attribute_name ILIKE 'data'`, but not all -- multiple attributes can contain JSON sensor payloads -- unless the user explicitly asks for a specific attribute, inspect all relevant JSON-carrying attributes that may hold telemetry for the chosen asset or group -- history is stored in asset datapoints, so historical queries should read from those datapoints directly -- do not use `ad.asset_id`; that column does not exist here -- always match the chosen asset using: - REPLACE(LOWER(a.name), ' ', '') = REPLACE(LOWER(''), ' ', '') -- OpenRemote time is already aligned with local time here, so do not apply a synthetic timezone offset -- keep the query to one statement only -- never write INSERT, UPDATE, DELETE, DROP, ALTER, or TRUNCATE -- return SQL only, no markdown fences and no explanation - -Use the asset context and the user request to decide: -- which telemetry keys to read -- whether the query should fetch latest rows or historical rows -- which computed columns are useful for later analysis -- whether the selected asset is a parent/group and therefore requires a recursive child-asset query using openremote.asset.parent_id -- prefer capturing telemetry across multiple JSON-carrying attributes rather than assuming only one attribute name - -Classification rule: -- an asset is a real device only when its attributes contain keys beyond notes and location -- if it is only a category/group, do not expect direct sensor telemetry from it -- for category/group questions, fetch the relevant child devices and their telemetry -- categories can contain sub-categories and those can contain devices, so recursive hierarchy queries are allowed -- if a recursive CTE is used, it must start with `WITH RECURSIVE` - -Tank rule: -- for tank devices, actual tank level must be computed as MaxFt - TankOilLevelInFeet001 -- the two source values may come from different attributes, so inspect all JSON-bearing attributes for `MaxFt`, `TankOilLevelInFeet001`, or `Level` -- if that computed level would be negative, clamp it to 0 - -If the selected asset has no direct live data but has children: -- use a recursive query or direct child query on openremote.asset.parent_id -- then join the child assets to openremote.asset_datapoint via ad.entity_id = child_asset.id -- summarize the relevant child assets for the user request - -The result should be accurate, safe, and directly executable. diff --git a/agents/analysis/backend/app/services/__init__.py b/agents/analysis/backend/app/services/__init__.py deleted file mode 100644 index d3f5a12faa99758192ecc4ed3fc22c9249232e86..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/__init__.py +++ /dev/null @@ -1 +0,0 @@ - diff --git a/agents/analysis/backend/app/services/__pycache__/__init__.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/__init__.cpython-313.pyc deleted file mode 100644 index cf6630b601f52ef60b62d68bd77240a0aa7a53ef..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/__init__.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/analysis_agent.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/analysis_agent.cpython-313.pyc deleted file mode 100644 index ddb315e799f01e5ae31a511a05553f1a2d82a573..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/analysis_agent.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/asset_analysis_instructions.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/asset_analysis_instructions.cpython-313.pyc deleted file mode 100644 index 39e1a6feb84ecb117e02e891caf072cf056bf97a..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/asset_analysis_instructions.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/asset_catalog.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/asset_catalog.cpython-313.pyc deleted file mode 100644 index 13640c802d34a70be3c6ed921d8845ac9ff90f7b..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/asset_catalog.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/data_repository.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/data_repository.cpython-313.pyc deleted file mode 100644 index 29919d47f7ee7a2a223c5122f4de819fbe497e7b..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/data_repository.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/deterministic_analysis.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/deterministic_analysis.cpython-313.pyc deleted file mode 100644 index 708977b1003871d00356e58a2b33e5e70ccdf0f9..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/deterministic_analysis.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/live_asset_registry.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/live_asset_registry.cpython-313.pyc deleted file mode 100644 index e6eeb5a2da459d395a3baea7026c9d7ff5f86744..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/live_asset_registry.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/llm_service.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/llm_service.cpython-313.pyc deleted file mode 100644 index 9495b6344c4dc96c3b18e246159aa4fe855f615b..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/llm_service.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/nl_to_sql.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/nl_to_sql.cpython-313.pyc deleted file mode 100644 index 6f5cd60d0095d42f763dff977463aa7d4f43a786..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/nl_to_sql.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/report_builder.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/report_builder.cpython-313.pyc deleted file mode 100644 index 0c7e5756bff467d25ed3bfa14f09dbb79b9c0ed9..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/report_builder.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/scheduling_agent.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/scheduling_agent.cpython-313.pyc deleted file mode 100644 index 4689f104e2f81870f8a486bd9eaa686949a30fbc..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/scheduling_agent.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/sql_guard.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/sql_guard.cpython-313.pyc deleted file mode 100644 index 47754fd18a97fe4124bcb26a92ca6ed0dd2837a4..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/sql_guard.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/__pycache__/trend_chart_service.cpython-313.pyc b/agents/analysis/backend/app/services/__pycache__/trend_chart_service.cpython-313.pyc deleted file mode 100644 index 7e546e99d4687577ff67e31a2128164b4c2316b0..0000000000000000000000000000000000000000 Binary files a/agents/analysis/backend/app/services/__pycache__/trend_chart_service.cpython-313.pyc and /dev/null differ diff --git a/agents/analysis/backend/app/services/analysis_agent.py b/agents/analysis/backend/app/services/analysis_agent.py deleted file mode 100644 index 11fe53f423786ed94ac4bab8e701b74bcf30b8e1..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/analysis_agent.py +++ /dev/null @@ -1,876 +0,0 @@ -from __future__ import annotations - -import json -import math -import re -from datetime import date, datetime, timedelta -from zoneinfo import ZoneInfo - -from fastapi import HTTPException - -from agents.analysis.backend.app.models.schemas import ( - AnalysisPlan, - AnalysisRequest, - AnalysisResponse, - ChatRequest, - ChatResponse, - TimeWindow, -) -from agents.analysis.backend.app.services.asset_analysis_instructions import build_asset_llm_context -from agents.analysis.backend.app.services.data_repository import DataQueryError, data_repository -from agents.analysis.backend.app.services.deterministic_analysis import deterministic_analysis_service -from agents.analysis.backend.app.services.live_asset_registry import live_asset_registry -from agents.analysis.backend.app.services.llm_service import llm_service -from agents.analysis.backend.app.services.nl_to_sql import nl_to_sql_service -from agents.analysis.backend.app.services.report_builder import report_builder_service -from agents.analysis.backend.app.services.sql_guard import validate_read_only_sql -from agents.analysis.backend.app.services.trend_chart_service import trend_chart_service - - -class AnalysisAgentService: - async def run_chat(self, payload: ChatRequest) -> ChatResponse: - from agents.analysis.backend.app.services.scheduling_agent import n8n_fast_path_service - - fast_path = n8n_fast_path_service.try_intercept(payload.message) - if fast_path: - return fast_path - - analysis_request = AnalysisRequest( - asset_id=payload.asset_id, - question=payload.message, - ) - analysis = await self.run_analysis(analysis_request, user_id=payload.user_id) - return ChatResponse(answer=analysis.report.summary, analysis=analysis) - - async def run_analysis(self, payload: AnalysisRequest, user_id: str = "user_1") -> AnalysisResponse: - candidates = live_asset_registry.find_candidates(payload.question, user_id=user_id) - direct_match = live_asset_registry.best_direct_match(payload.question, user_id=user_id) - planner_result = self._default_planner_result() - if direct_match is not None: - planner_result["asset_id"] = direct_match.asset_id - planner_result["reasoning"] = "Direct asset match found from the user message." - elif payload.asset_id: - planner_result["asset_id"] = payload.asset_id - planner_result["reasoning"] = "Asset id was provided directly in request." - else: - planner_result = llm_service.plan_request( - user_question=payload.question, - asset_catalog_context=self._build_asset_catalog_context(candidates), - ) - - asset = self._resolve_asset(payload.asset_id, planner_result, payload.question, candidates, user_id=user_id) - - # Guardrail: If user asked for 'coldroom' but we picked a 'tank', check if we have better candidates - if asset.asset_type == "tank" and "cold" in payload.question.lower() and "room" in payload.question.lower(): - for c in candidates: - if c.asset_type == "cold_room": - asset = c - planner_result["reasoning"] = "User asked for coldrooms, overriding tank selection with a valid coldroom candidate." - break - - analysis_name = payload.analysis_name or planner_result.get("analysis_name") or "historical_variance" - time_window = payload.time_window or self._parse_time_window(planner_result.get("time_window")) - explicit_time_window = self._extract_explicit_time_range_window(payload.question) - if explicit_time_window is not None: - started_at, ended_at, label = explicit_time_window - duration_hours = max( - 1, - int(math.ceil((ended_at - started_at).total_seconds() / 3600)), - ) - time_window = TimeWindow( - scope="historical", - value=duration_hours, - unit="hours", - label=label, - start_at=started_at.isoformat(), - end_at=ended_at.isoformat(), - ) - elif (relative_day_window := self._extract_relative_day_window(payload.question)) is not None: - started_at, ended_at, label = relative_day_window - duration_hours = max( - 1, - int(math.ceil((ended_at - started_at).total_seconds() / 3600)), - ) - time_window = TimeWindow( - scope="historical", - value=duration_hours, - unit="hours", - label=label, - start_at=started_at.isoformat(), - end_at=ended_at.isoformat(), - ) - elif (relative_duration_window := self._extract_relative_duration_window(payload.question)) is not None: - time_window = relative_duration_window - if explicit_time_window is None and (specific_date_window := self._extract_specific_date_window(payload.question)) is not None: - started_at, ended_at, label = specific_date_window - time_window = TimeWindow( - scope="historical", - value=1, - unit="days", - label=label, - start_at=started_at.isoformat(), - end_at=ended_at.isoformat(), - ) - if payload.time_window is None and not self._question_mentions_time_window(payload.question): - time_window = self._default_analysis_time_window() - asset_context = build_asset_llm_context(asset) - query_window = self._build_query_window(time_window) - - use_deterministic_sql = ( - asset.asset_type in {"tank", "cold_room", "smoke_alarm"} - or (not asset.is_device and asset.child_count > 0) - ) - sql_result = {"sql_query": "", "total_tokens": 0, "raw_response": ""} - if use_deterministic_sql: - sql_query = nl_to_sql_service.build_query( - asset=asset, - question=payload.question, - analysis_name=analysis_name, - time_window=time_window, - ).query - else: - sql_result = llm_service.generate_sql( - user_question=payload.question, - asset_context=asset_context, - analysis_name=analysis_name, - time_window_label=time_window.label, - ) - sql_query = nl_to_sql_service.normalize_generated_query(sql_result["sql_query"], asset) - - warnings: list[str] = [] - try: - validate_read_only_sql(sql_query) - except Exception: - fallback_sql = nl_to_sql_service.build_query( - asset=asset, - question=payload.question, - analysis_name=analysis_name, - time_window=time_window, - ) - sql_query = fallback_sql.query - warnings.append("The LLM SQL was invalid, so a safe fallback SQL template was used.") - - try: - rows = data_repository.execute_query(sql_query) - except DataQueryError as exc: - fallback_sql = nl_to_sql_service.build_query( - asset=asset, - question=payload.question, - analysis_name=analysis_name, - time_window=time_window, - ) - sql_query = fallback_sql.query - warnings.append("The generated SQL failed at runtime, so a safe fallback query was used.") - try: - rows = data_repository.execute_query(sql_query) - except DataQueryError as retry_exc: - raise HTTPException(status_code=502, detail=str(retry_exc)) from retry_exc - rows = self._clip_rows_to_window(rows, time_window) - # Flatten rows to ensure nested sensor_data is usable by charting and deterministic tools - flattened_rows = [] - for row in rows: - new_row = dict(row) - sd = row.get("sensor_data") - if isinstance(sd, dict): - for k, v in sd.items(): - if k not in new_row: - new_row[k] = v - flattened_rows.append(new_row) - rows = flattened_rows - - latest_row_time = self._latest_row_timestamp(rows) - query_window = self._apply_display_window_override(payload.question, time_window, query_window) - - deterministic_summary = deterministic_analysis_service.summarize(asset, rows) - warnings.extend(list(deterministic_summary["warnings"])) - if latest_row_time is not None: - try: - request_end = datetime.fromisoformat(query_window["ended_at_iso"]) - if latest_row_time.tzinfo is None and request_end.tzinfo is not None: - latest_row_time = latest_row_time.replace(tzinfo=request_end.tzinfo) - if latest_row_time < request_end: - diff = request_end - latest_row_time - if diff.total_seconds() > 3600: - hours = int(diff.total_seconds() // 3600) - warnings.append( - f"The data for {asset.name} appears to be stale ({hours} hours old). " - f"Latest telemetry available is from {latest_row_time.strftime('%d %b %Y, %I:%M %p')} (Asia/Karachi)." - ) - else: - warnings.append( - f"Latest telemetry available for {asset.name} in this window is {latest_row_time.strftime('%d %b %Y, %I:%M %p')} (Asia/Karachi)." - ) - except Exception: - pass - - if asset.asset_type == "smoke_alarm": - final_analysis = { - "answer": self._build_smoke_alarm_answer(asset, time_window, deterministic_summary, rows), - "total_tokens": 0, - "raw_response": "", - "prompt_text": "", - } - else: - final_analysis = llm_service.analyze_result( - user_question=payload.question, - asset_context=asset_context, - sql_query=sql_query, - deterministic_findings=deterministic_summary, - rows=rows, - ) - - total_tokens = ( - int(sql_result.get("total_tokens", 0)) - + int(final_analysis.get("total_tokens", 0)) - ) - - plan = AnalysisPlan( - analysis_name=analysis_name, - reasoning=planner_result.get("reasoning", ""), - time_window=time_window, - query_started_at=query_window["started_at_iso"], - query_ended_at=query_window["ended_at_iso"], - query_window_label=query_window["label"], - planner_prompt=planner_result.get("prompt_text", ""), - sql_prompt=sql_result.get("prompt_text", ""), - analyst_prompt=final_analysis.get("prompt_text", ""), - sql_query=sql_query, - chart_hints=self._chart_hints_for_analysis(analysis_name), - warnings=warnings, - total_tokens=total_tokens, - llm_raw_response=final_analysis.get("raw_response", ""), - ) - report = report_builder_service.build( - asset=asset, - plan=plan, - rows=rows, - summary_override=final_analysis.get("answer", ""), - ) - trend_chart = trend_chart_service.build(asset, analysis_name, time_window, rows) - trend_charts = [trend_chart] if trend_chart else [] - return AnalysisResponse( - asset=asset, - plan=plan, - report=report, - trend_chart=trend_chart, - trend_charts=trend_charts, - rows=rows, - ) - - def _build_smoke_alarm_answer( - self, - asset, - time_window: TimeWindow, - deterministic_summary: dict[str, object], - rows: list[dict], - ) -> str: - findings = [str(item).strip() for item in deterministic_summary.get("findings", []) if str(item).strip()] - warnings = [str(item).strip() for item in deterministic_summary.get("warnings", []) if str(item).strip()] - - if not rows: - lines = [ - f"No smoke-alarm telemetry rows were found for {asset.name} in {time_window.label}.", - ] - if warnings: - lines.append("") - lines.extend(f"- {item}" for item in warnings[:4]) - return "\n".join(lines) - - headline = findings[0] if findings else f"{asset.name} smoke-alarm telemetry was analyzed for {time_window.label}." - latest_states = [item for item in findings[1:] if ": " in item][:6] - other_findings = [item for item in findings[1:] if item not in latest_states][:4] - - lines = [headline] - - if other_findings: - lines.append("") - lines.append("Key points") - lines.extend(f"- {item}" for item in other_findings) - - if latest_states: - lines.append("") - lines.append("Latest device states") - lines.extend(f"- {item}" for item in latest_states) - - if warnings: - lines.append("") - lines.append("Attention items") - lines.extend(f"- {item}" for item in warnings[:4]) - - return "\n".join(lines) - - def _apply_display_window_override( - self, - question: str, - time_window: TimeWindow, - query_window: dict[str, str], - ) -> dict[str, str]: - text = question.lower() - # Keep window as-is unless user asked for "today". - if "today" not in text: - return query_window - if not time_window.start_at: - return query_window - start_at = self._parse_datetime(time_window.start_at) - if start_at is None: - return query_window - end_at = self._parse_datetime(query_window.get("ended_at_iso", "")) - if end_at is None: - return query_window - - friendly_format = "%d %b %Y, %I:%M %p" - query_window["label"] = ( - f"{start_at.strftime(friendly_format)} to " - f"{end_at.strftime(friendly_format)} " - f"(Asia/Karachi)" - ) - return query_window - - def _latest_row_timestamp(self, rows: list[dict]) -> datetime | None: - timestamp_keys = ("recorded_at", "timestamp", "local_time", "raw_time", "latest_time") - latest: datetime | None = None - for row in rows: - for key in timestamp_keys: - value = row.get(key) - if value is None: - continue - parsed = self._parse_datetime(value) - if parsed is None: - continue - if latest is None or parsed > latest: - latest = parsed - break - return latest - - def _resolve_asset( - self, - requested_asset_id: str | None, - planner_result: dict, - question: str, - candidates: list, - user_id: str = "user_1", - ) -> object: - if requested_asset_id: - asset = live_asset_registry.get_asset(requested_asset_id, user_id=user_id) - if asset is not None: - return asset - - direct_match = live_asset_registry.best_direct_match(question, user_id=user_id) - if direct_match is not None: - return direct_match - - planned_asset_id = planner_result.get("asset_id", "") - if planned_asset_id: - asset = live_asset_registry.get_asset(planned_asset_id, user_id=user_id) - if asset is not None: - return asset - - available = list(dict.fromkeys(asset.name for asset in (candidates[:12] or live_asset_registry.list_assets(user_id)[:12]))) - detail = "I could not identify the asset from your message." - if available: - detail += f" Available assets include: {', '.join(available)}." - raise HTTPException(status_code=422, detail=detail) - - def _parse_time_window(self, raw_time_window: dict | None) -> TimeWindow: - if not raw_time_window: - return self._default_analysis_time_window() - try: - return TimeWindow(**raw_time_window) - except Exception: - return self._default_analysis_time_window() - - def _default_analysis_time_window(self) -> TimeWindow: - return TimeWindow(scope="historical", value=3, unit="hours", label="last 3 hours") - - def _question_mentions_time_window(self, question: str) -> bool: - text = question.lower() - if self._extract_explicit_time_range_window(question) is not None: - return True - if self._extract_relative_duration_window(question) is not None: - return True - if self._extract_relative_day_window(question) is not None: - return True - if self._extract_specific_date_window(question) is not None: - return True - - time_words = ( - "today", - "yesterday", - "tomorrow", - "last", - "past", - "previous", - "between", - "from", - "to", - "since", - "until", - "till", - "hour", - "hours", - "minute", - "minutes", - "day", - "days", - "week", - "weeks", - "month", - "months", - ) - return any(word in text for word in time_words) - - def _build_query_window(self, time_window: TimeWindow) -> dict[str, str]: - timezone = ZoneInfo("Asia/Karachi") - if time_window.start_at and time_window.end_at: - started_at = datetime.fromisoformat(time_window.start_at) - ended_at = datetime.fromisoformat(time_window.end_at) - friendly_format = "%d %b %Y, %I:%M %p" - return { - "started_at_iso": started_at.isoformat(), - "ended_at_iso": ended_at.isoformat(), - "label": ( - f"{started_at.strftime(friendly_format)} to " - f"{ended_at.strftime(friendly_format)} (Asia/Karachi)" - ), - } - - ended_at = datetime.now(timezone) - - unit_to_delta = { - "minutes": timedelta(minutes=time_window.value), - "hours": timedelta(hours=time_window.value), - "days": timedelta(days=time_window.value), - "weeks": timedelta(weeks=time_window.value), - "months": timedelta(days=30 * time_window.value), - } - started_at = ended_at - unit_to_delta.get(time_window.unit, timedelta(hours=3)) - friendly_format = "%d %b %Y, %I:%M %p" - - return { - "started_at_iso": started_at.isoformat(), - "ended_at_iso": ended_at.isoformat(), - "label": ( - f"{started_at.strftime(friendly_format)} to " - f"{ended_at.strftime(friendly_format)} (Asia/Karachi)" - ), - } - - def _extract_specific_date_window(self, question: str) -> tuple[datetime, datetime, str] | None: - timezone = ZoneInfo("Asia/Karachi") - patterns = [ - r"\b(\d{4})-(\d{1,2})-(\d{1,2})\b", - r"\b(\d{1,2})/(\d{1,2})/(\d{4})\b", - r"\b(\d{1,2})-(\d{1,2})-(\d{4})\b", - ] - - for pattern in patterns: - match = re.search(pattern, question) - if not match: - continue - groups = match.groups() - try: - if pattern.startswith(r"\b(\d{4})"): - year, month, day = int(groups[0]), int(groups[1]), int(groups[2]) - else: - day, month, year = int(groups[0]), int(groups[1]), int(groups[2]) - anchor_date = datetime(year, month, day, 6, 0, 0, tzinfo=timezone) - end_date = anchor_date + timedelta(days=1) - label = ( - f"{anchor_date.strftime('%d %b %Y')} 06:00 AM to " - f"{end_date.strftime('%d %b %Y')} 06:00 AM" - ) - return anchor_date, end_date, label - except ValueError: - continue - - month_name_patterns = [ - r"\b(\d{1,2})(?:st|nd|rd|th)?\s+([A-Za-z]+)\s*,?\s*(\d{4})\b", - r"\b([A-Za-z]+)\s+(\d{1,2})(?:st|nd|rd|th)?\s*,?\s*(\d{4})\b", - r"\b(\d{1,2})(?:st|nd|rd|th)?\s+([A-Za-z]+)\b", - r"\b([A-Za-z]+)\s+(\d{1,2})(?:st|nd|rd|th)?\b", - ] - month_map = { - "jan": 1, "january": 1, - "feb": 2, "february": 2, - "mar": 3, "march": 3, - "apr": 4, "april": 4, - "may": 5, - "jun": 6, "june": 6, - "jul": 7, "july": 7, - "aug": 8, "august": 8, - "sep": 9, "sept": 9, "september": 9, - "oct": 10, "october": 10, - "nov": 11, "november": 11, - "dec": 12, "decebmber": 12, "december": 12, - } - now_local = datetime.now(timezone) - for pattern in month_name_patterns: - match = re.search(pattern, question, flags=re.IGNORECASE) - if not match: - continue - groups = match.groups() - try: - if pattern.startswith(r"\b(\d"): - day = int(groups[0]) - month = month_map.get(groups[1].lower()) - year = int(groups[2]) if len(groups) > 2 and groups[2] else now_local.year - else: - month = month_map.get(groups[0].lower()) - day = int(groups[1]) - year = int(groups[2]) if len(groups) > 2 and groups[2] else now_local.year - if month is None: - continue - anchor_date = datetime(year, month, day, 6, 0, 0, tzinfo=timezone) - end_date = anchor_date + timedelta(days=1) - label = ( - f"{anchor_date.strftime('%d %b %Y')} 06:00 AM to " - f"{end_date.strftime('%d %b %Y')} 06:00 AM" - ) - return anchor_date, end_date, label - except ValueError: - continue - - ordinal_only = re.search(r"\b(\d{1,2})(?:st|nd|rd|th)\b", question, flags=re.IGNORECASE) - if ordinal_only: - day = int(ordinal_only.group(1)) - year = now_local.year - month = now_local.month - if day > now_local.day: - if month == 1: - month = 12 - year -= 1 - else: - month -= 1 - try: - anchor_date = datetime(year, month, day, 6, 0, 0, tzinfo=timezone) - end_date = anchor_date + timedelta(days=1) - label = ( - f"{anchor_date.strftime('%d %b %Y')} 06:00 AM to " - f"{end_date.strftime('%d %b %Y')} 06:00 AM" - ) - return anchor_date, end_date, label - except ValueError: - return None - return None - - def _extract_relative_duration_window(self, question: str) -> TimeWindow | None: - lowered = question.lower() - - unit_map = { - "h": "hours", - "hr": "hours", - "hrs": "hours", - "hour": "hours", - "hours": "hours", - "d": "days", - "day": "days", - "days": "days", - "w": "weeks", - "wk": "weeks", - "wks": "weeks", - "week": "weeks", - "weeks": "weeks", - "m": "months", - "mo": "months", - "mon": "months", - "month": "months", - "months": "months", - "min": "minutes", - "mins": "minutes", - "minute": "minutes", - "minutes": "minutes", - } - - def build_window(value: int, raw_unit: str) -> TimeWindow | None: - unit = unit_map.get(raw_unit.lower()) - if unit is None: - return None - return TimeWindow( - scope="historical", - value=value, - unit=unit, - label=f"last {value} {unit}", - ) - - pattern = r"\b(?:last|past|for|previous|in)\s+(\d+)\s*(h|hr|hrs?|hours?|d|days?|w|wk|wks|weeks?|m|mo|mon|months?|min|mins?|minutes?)\b" - match = re.search(pattern, lowered) - if match: - built = build_window(int(match.group(1)), match.group(2)) - if built is not None: - return built - - pattern_single = r"\b(?:last|past|for|previous|in)\s+(?:an?|1)\s+(h|hr|hrs?|hours?|d|days?|w|wk|wks|weeks?|m|mo|mon|months?|min|mins?|minutes?)\b" - match_single = re.search(pattern_single, lowered) - if match_single: - built = build_window(1, match_single.group(1)) - if built is not None: - return built - - pattern_bare = r"\b(\d+)\s*(h|hr|hrs?|hours?|d|days?|w|wk|wks|weeks?|m|mo|mon|months?|min|mins?|minutes?)\b" - match_bare = re.search(pattern_bare, lowered) - if match_bare: - built = build_window(int(match_bare.group(1)), match_bare.group(2)) - if built is not None: - return built - - return None - - def _extract_explicit_time_range_window(self, question: str) -> tuple[datetime, datetime, str] | None: - timezone = ZoneInfo("Asia/Karachi") - lowered = question.lower() - range_match = re.search( - r"\b(?:from\s+)?(\d{1,2}(?::\d{2})?\s*(?:am|pm))\s*(?:to|-|till|until)\s*(\d{1,2}(?::\d{2})?\s*(?:am|pm))\b", - lowered, - flags=re.IGNORECASE, - ) - if not range_match: - return None - - start_hm = self._parse_clock_token(range_match.group(1)) - end_hm = self._parse_clock_token(range_match.group(2)) - if start_hm is None or end_hm is None: - return None - - anchor = self._extract_anchor_date(question, timezone) or datetime.now(timezone).date() - started_at = datetime(anchor.year, anchor.month, anchor.day, start_hm[0], start_hm[1], tzinfo=timezone) - ended_at = datetime(anchor.year, anchor.month, anchor.day, end_hm[0], end_hm[1], tzinfo=timezone) - if ended_at <= started_at: - ended_at += timedelta(days=1) - - friendly_format = "%d %b %Y, %I:%M %p" - label = f"{started_at.strftime(friendly_format)} to {ended_at.strftime(friendly_format)} (Asia/Karachi)" - return started_at, ended_at, label - - def _parse_clock_token(self, token: str) -> tuple[int, int] | None: - match = re.match(r"^\s*(\d{1,2})(?::(\d{2}))?\s*(am|pm)\s*$", token.strip(), flags=re.IGNORECASE) - if not match: - return None - hour = int(match.group(1)) - minute = int(match.group(2) or "0") - meridian = match.group(3).lower() - if hour < 1 or hour > 12 or minute < 0 or minute > 59: - return None - if meridian == "am": - hour = 0 if hour == 12 else hour - else: - hour = 12 if hour == 12 else hour + 12 - return hour, minute - - def _extract_anchor_date(self, question: str, timezone: ZoneInfo) -> date | None: - patterns = [ - r"\b(\d{4})-(\d{1,2})-(\d{1,2})\b", - r"\b(\d{1,2})/(\d{1,2})/(\d{4})\b", - r"\b(\d{1,2})-(\d{1,2})-(\d{4})\b", - ] - - for pattern in patterns: - match = re.search(pattern, question) - if not match: - continue - groups = match.groups() - try: - if pattern.startswith(r"\b(\d{4})"): - year, month, day = int(groups[0]), int(groups[1]), int(groups[2]) - else: - day, month, year = int(groups[0]), int(groups[1]), int(groups[2]) - return date(year, month, day) - except ValueError: - continue - - month_name_patterns = [ - r"\b(\d{1,2})(?:st|nd|rd|th)?\s+([A-Za-z]+)\s*,?\s*(\d{4})\b", - r"\b([A-Za-z]+)\s+(\d{1,2})(?:st|nd|rd|th)?\s*,?\s*(\d{4})\b", - r"\b(\d{1,2})(?:st|nd|rd|th)?\s+([A-Za-z]+)\b", - r"\b([A-Za-z]+)\s+(\d{1,2})(?:st|nd|rd|th)?\b", - ] - month_map = { - "jan": 1, "january": 1, - "feb": 2, "february": 2, - "mar": 3, "march": 3, - "apr": 4, "april": 4, - "may": 5, - "jun": 6, "june": 6, - "jul": 7, "july": 7, - "aug": 8, "august": 8, - "sep": 9, "sept": 9, "september": 9, - "oct": 10, "october": 10, - "nov": 11, "november": 11, - "dec": 12, "decebmber": 12, "december": 12, - } - now_local = datetime.now(timezone) - for pattern in month_name_patterns: - match = re.search(pattern, question, flags=re.IGNORECASE) - if not match: - continue - groups = match.groups() - try: - if pattern.startswith(r"\b(\d"): - day = int(groups[0]) - month = month_map.get(groups[1].lower()) - year = int(groups[2]) if len(groups) > 2 and groups[2] else now_local.year - else: - month = month_map.get(groups[0].lower()) - day = int(groups[1]) - year = int(groups[2]) if len(groups) > 2 and groups[2] else now_local.year - if month is None: - continue - return date(year, month, day) - except ValueError: - continue - - ordinal_only = re.search(r"\b(\d{1,2})(?:st|nd|rd|th)\b", question, flags=re.IGNORECASE) - if ordinal_only: - day = int(ordinal_only.group(1)) - year = now_local.year - month = now_local.month - if day > now_local.day: - if month == 1: - month = 12 - year -= 1 - else: - month -= 1 - try: - return date(year, month, day) - except ValueError: - return None - return None - - def _extract_relative_day_window(self, question: str) -> tuple[datetime, datetime, str] | None: - text = question.lower() - timezone = ZoneInfo("Asia/Karachi") - now_local = datetime.now(timezone) - - if "today" in text or "scheduled" in text: - shift_start = now_local.replace(hour=6, minute=0, second=0, microsecond=0) - if now_local < shift_start: - shift_start = shift_start - timedelta(days=1) - label = ( - f"{shift_start.strftime('%d %b %Y')} 06:00 AM to " - f"{now_local.strftime('%d %b %Y %I:%M %p')}" - ) - return shift_start, now_local, label - - if "yesterday" in text: - today_shift_start = now_local.replace(hour=6, minute=0, second=0, microsecond=0) - if now_local < today_shift_start: - today_shift_start = today_shift_start - timedelta(days=1) - yesterday_shift_start = today_shift_start - timedelta(days=1) - label = ( - f"{yesterday_shift_start.strftime('%d %b %Y')} 06:00 AM to " - f"{today_shift_start.strftime('%d %b %Y')} 06:00 AM" - ) - return yesterday_shift_start, today_shift_start, label - - return None - - def _clip_rows_to_window(self, rows: list[dict], time_window: TimeWindow) -> list[dict]: - if not rows or not time_window.start_at or not time_window.end_at: - return rows - try: - start_dt = datetime.fromisoformat(time_window.start_at) - end_dt = datetime.fromisoformat(time_window.end_at) - except ValueError: - return rows - - timestamp_keys = ("recorded_at", "timestamp", "local_time", "raw_time") - filtered: list[dict] = [] - for row in rows: - row_ts = None - for key in timestamp_keys: - value = row.get(key) - if value is None: - continue - row_ts = self._parse_datetime(value) - if row_ts is not None: - break - if row_ts is None: - filtered.append(row) - continue - - if row_ts.tzinfo is None and start_dt.tzinfo is not None: - row_ts = row_ts.replace(tzinfo=start_dt.tzinfo) - if start_dt <= row_ts < end_dt: - filtered.append(row) - return filtered - - def _parse_datetime(self, value: object) -> datetime | None: - if isinstance(value, datetime): - return value - text = str(value).strip() - if not text: - return None - try: - return datetime.fromisoformat(text.replace("Z", "+00:00")) - except ValueError: - pass - for fmt in ( - "%d %b %Y, %I:%M:%S %p", - "%d %b %Y, %I:%M %p", - "%d-%m-%Y %I:%M %p", - "%Y-%m-%d %H:%M:%S", - "%Y-%m-%d %H:%M", - ): - try: - return datetime.strptime(text, fmt) - except ValueError: - continue - return None - - def _default_planner_result(self) -> dict: - return { - "asset_id": "", - "analysis_name": "live_status", - "time_window": { - "scope": "historical", - "value": 3, - "unit": "hours", - "label": "last 3 hours", - }, - "reasoning": "Planner bypassed.", - "prompt_text": "[BYPASSED]", - } - - def _build_asset_catalog_context(self, candidates: list) -> str: - shortlisted = candidates[:4] if candidates else live_asset_registry.list_assets()[:5] - entries: list[str] = [] - total_chars = 0 - max_chars = 2400 - for asset in shortlisted: - aliases = [ - alias for alias in asset.lookup_names - if alias and alias.lower() != asset.name.lower() - ][:2] - entry = json.dumps( - { - "asset_id": asset.asset_id, - "name": asset.name, - "aliases": aliases, - "type": asset.asset_type, - "parent": asset.parent_name or "", - "child_count": asset.child_count, - "has_live_data": asset.has_live_data, - "data_keys": asset.data_keys[:4], - "supported_analyses": asset.supported_analyses[:3], - } - ) - if total_chars + len(entry) > max_chars: - break - entries.append(entry) - total_chars += len(entry) - return "\n".join(entries) - - def _chart_hints_for_analysis(self, analysis_name: str) -> list[str]: - mapping = { - "cooling_cycle_analysis": ["line_chart", "event_markers"], - "consumption_analysis": ["line_chart", "duration_summary", "event_markers"], - "live_status": ["stat_cards"], - "trend_analysis": ["line_chart"], - "historical_variance": ["line_chart", "box_plot"], - "throughput_utilization": ["bar_chart", "line_chart"], - "inventory_level_analysis": ["line_chart", "threshold_bands"], - "anomaly_detection": ["line_chart", "anomaly_table"], - } - return mapping.get(analysis_name, ["line_chart"]) - - -analysis_agent_service = AnalysisAgentService() diff --git a/agents/analysis/backend/app/services/asset_analysis_instructions.py b/agents/analysis/backend/app/services/asset_analysis_instructions.py deleted file mode 100644 index 791d0f52a2bac4ed0fb489f9bf61ca83d1d1e8d1..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/asset_analysis_instructions.py +++ /dev/null @@ -1,178 +0,0 @@ -from __future__ import annotations - -from agents.analysis.backend.app.core.config import get_settings -from agents.analysis.backend.app.models.schemas import AssetProfile - - -ASSET_TYPE_ANALYSIS_GUIDANCE = { - "cold_room": """ -For cold rooms, prioritize: -- live device health and payload freshness -- current temperature and humidity -- acceptable temperature band compliance -- historical drift, highs, lows, and instability -- cooling cycle estimation and repeated threshold breaches -- consumption analysis from temperature history when the asset belongs to T & H Monitoring - -Preferred analysis choice: -- if user asks about live/current/status -> live_status -- if the asset is in T & H Monitoring and the user asks for a time window, prefer consumption_analysis -- if user asks about temperature pattern, cooling, stability -> cooling_cycle_analysis -- if user asks for previous day/week/month comparison -> historical_variance -- if user asks about abnormal behavior -> anomaly_detection -""".strip(), - "filling_machine": """ -For filling machines, prioritize: -- current production counters -- per-shift throughput -- stale payload detection -- production movement over the requested time window -- identifying flat output, drops, or sudden spikes - -Preferred analysis choice: -- if user asks about live/current/status -> live_status -- if user asks about production, throughput, utilization -> throughput_utilization -- if user asks about previous windows or comparisons -> historical_variance -- if user asks about unusual behavior -> anomaly_detection -""".strip(), - "tank": """ -For tanks, prioritize: -- current level -- short-term and long-term rate of change -- refill and depletion pattern -- sharp drops and low-level risk -- stale payload detection - -Preferred analysis choice: -- if user asks about live/current/status -> live_status -- if user asks about level movement or stock behavior -> inventory_level_analysis -- if user asks for historical movement -> trend_analysis -- if user asks about unusual drops -> anomaly_detection -""".strip(), - "smoke_alarm": """ -For smoke alarm systems, prioritize: -- Enlisting ALL assets and ALL their attributes in a concise summary. -- Use the 'static_attributes' JSON provided in the data rows to discover all configured attributes for an asset, even if they have no live telemetry. -- If all are okay, keep the per-asset summary brief (one line per asset). -- If smoke, fire, fault, or low-battery is detected, provide the detailed breakdown immediately for that specific asset. -- Highlight smoke/fire detection events prominently. -- Ensure health status, battery, temperature, and humidity are mentioned for every device. - -Preferred analysis choice: -- if user asks about live/current/status -> live_status -- if user asks whether smoke/fire happened -> anomaly_detection -- if user asks about a category/group -> group_status -""".strip(), - "energy_meter": """ -For energy meters, prioritize: -- overall power and energy consumption -- peak usage identification -- breaking down consumption by child meters -- power factor anomalies -- voltage or current imbalances - -Preferred analysis choice: -- if user asks about live/current/status -> live_status -- if user asks about consumption, peak usage, or breakdown -> consumption_analysis -- if user asks about historical trends -> trend_analysis -- if user asks about a category/group -> group_status -""".strip(), -} - - -ASSET_NAME_OVERRIDES = { - "Cold Room 1": """ -This cold room should be treated as a temperature-preservation asset. -The final answer should clearly mention whether it is stable, outside band, -or stale/offline, and whether temperature movement suggests a cooling issue. -""".strip(), - "Filling Machine 1": """ -This machine should be treated as a throughput asset. -The final answer should clearly mention total count movement, strongest shift, -and whether the payload looks stale or flat. -""".strip(), - "Oil Tank 1": """ -This tank should be treated as an inventory asset. -The final answer should clearly mention current level, whether the trend is falling, -and whether the level looks risky. -""".strip(), -} - - -def build_asset_llm_context(asset: AssetProfile) -> str: - parts = [ - f"Asset selected from chat: {asset.name}", - f"Asset type: {asset.asset_type}", - f"Is device with sensor data: {asset.is_device}", - f"Parent asset: {asset.parent_name or 'None'}", - f"Direct child count: {asset.child_count}", - f"Asset description: {asset.description}", - f"Lookup aliases: {', '.join(asset.lookup_names) if asset.lookup_names else asset.name}", - f"Supported analyses: {', '.join(asset.supported_analyses)}", - f"Important metrics: {', '.join(asset.metrics)}", - f"Attribute keys: {', '.join(asset.attribute_keys[:20]) if asset.attribute_keys else 'None'}", - ( - "Exact telemetry keys: " - f"temperature={asset.temperature_keys}, " - f"humidity={asset.humidity_keys}, " - f"level={asset.level_keys}, " - f"production={asset.production_keys}" - ), - ( - "OpenRemote SQL facts: " - "join openremote.asset_datapoint to openremote.asset using ad.entity_id = a.id, " - "telemetry is usually under ad.value JSON, " - "multiple attributes can contain JSON sensor payloads, " - "the analysis should inspect all relevant JSON-carrying attributes unless the user narrows the request, " - "historical data should be read from asset datapoints, " - f"and OpenRemote time should be adjusted to local time by adding the {get_settings().local_time_offset_hours}-hour offset (e.g. ad.timestamp + INTERVAL '{get_settings().local_time_offset_hours} hours')." - ), - ( - "Device vs category rule: only assets whose attributes contain keys other than notes and location " - "should be treated as real sensor devices. Parent/group assets should be analyzed via their child assets." - ), - ( - f"Nominal thresholds: {asset.nominal_range if asset.nominal_range else {'min_temperature': 5.0, 'max_temperature': 45.0}}. " - "Use these thresholds for compliance checks and avoid introducing any unstated default range." - ), - ] - - if asset.asset_type == "tank": - parts.append( - "Tank rule: for tank-related devices, the actual physical liquid height is MaxFt - TankOilLevelInFeet001. " - "These values can come from two different attributes, so the analysis must combine them before drawing conclusions. " - "The final analysis should prioritize showing the percentage filled (e.g., 75% filled) rather than just the height, " - "using the formula (liquid height / MaxFt) * 100. " - "Do not explain how you calculated the level or percentage to the user." - ) - - if "t & h monitoring" in f"{asset.name} {asset.parent_name}".lower(): - parts.append( - "T & H Monitoring consumption rule: determine cold-room ON/OFF state from temperature trend rather than a direct signal. " - "A cooling cycle starts at the last peak before temperature begins decreasing and later drops below 20C. " - "The cycle stays ON while temperature remains below 20C and ends at the valley before temperature rises back to 20C or above." - "Sum all such cycle durations across the requested window, clip any cycle that began before the shift to 6:00 AM, " - "and if a cycle is still active at the end of the data, count it up to the latest timestamp." - ) - - if asset.asset_type == "smoke_alarm": - parts.append( - "Smoke-alarm rule: inspect every relevant JSON-carrying attribute for each end device. " - "Treat warn='warn' as smoke/fire detected, warn='mute' as alarm was muted after detection, " - "warn='fault' as device fault, warn='remove' as device removed, warn='install' as device installed, " - "warn='low-vol' or warn='low-vol-test' as low battery/voltage, and warn='ok' as normal. " - "The answer must clearly state whether smoke/fire was detected and include the timestamps of those events." - ) - - type_guidance = ASSET_TYPE_ANALYSIS_GUIDANCE.get(asset.asset_type) - if type_guidance: - parts.append(type_guidance) - - override = ASSET_NAME_OVERRIDES.get(asset.name) - if override: - parts.append(override) - - if asset.analysis_instructions: - parts.append(f"Asset-specific rules: {asset.analysis_instructions}") - - return "\n\n".join(parts) diff --git a/agents/analysis/backend/app/services/asset_catalog.py b/agents/analysis/backend/app/services/asset_catalog.py deleted file mode 100644 index 92689da81e7f0d510c350c64de24d346e1d24e14..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/asset_catalog.py +++ /dev/null @@ -1,65 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -import yaml - -from agents.analysis.backend.app.models.schemas import AssetProfile - - -class AssetCatalogService: - def __init__(self) -> None: - self._catalog_path = Path(__file__).resolve().parents[2] / "data" / "asset_catalog.yaml" - self._cache: dict[str, AssetProfile] | None = None - - def _load(self) -> dict[str, AssetProfile]: - raw = yaml.safe_load(self._catalog_path.read_text(encoding="utf-8")) - assets = raw.get("assets", []) - return { - asset["asset_id"]: AssetProfile(**asset) - for asset in assets - } - - def list_assets(self) -> list[AssetProfile]: - self._cache = self._load() - return list(self._cache.values()) - - def get_asset(self, asset_id: str) -> AssetProfile | None: - self._cache = self._load() - return self._cache.get(asset_id) - - def get_asset_by_name(self, name: str) -> AssetProfile | None: - normalized_name = self._normalize(name) - for asset in self.list_assets(): - aliases = [asset.name, asset.asset_id, *asset.lookup_names] - if any(self._normalize(alias) == normalized_name for alias in aliases): - return asset - return None - - def find_asset_by_message(self, message: str) -> AssetProfile | None: - normalized_message = self._normalize(message) - best_match: AssetProfile | None = None - best_length = -1 - - for asset in self.list_assets(): - aliases = [asset.asset_id, asset.name, *asset.lookup_names] - for alias in aliases: - normalized_alias = self._normalize(alias) - if normalized_alias and normalized_alias in normalized_message: - if len(normalized_alias) > best_length: - best_match = asset - best_length = len(normalized_alias) - return best_match - - def list_asset_names(self) -> list[str]: - names: list[str] = [] - for asset in self.list_assets(): - names.append(asset.name) - names.extend(asset.lookup_names) - return sorted(set(names)) - - def _normalize(self, value: str) -> str: - return "".join(value.lower().split()) - - -asset_catalog_service = AssetCatalogService() diff --git a/agents/analysis/backend/app/services/data_repository.py b/agents/analysis/backend/app/services/data_repository.py deleted file mode 100644 index c7ee23e909dcbc6068c48b8ea4b0168ce9517370..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/data_repository.py +++ /dev/null @@ -1,49 +0,0 @@ -from __future__ import annotations - -from datetime import date, datetime - -from sqlalchemy import create_engine, text -from sqlalchemy.exc import SQLAlchemyError - -from agents.analysis.backend.app.core.config import get_settings - - -class DataQueryError(RuntimeError): - pass - - -class DataRepository: - def __init__(self) -> None: - self._settings = get_settings() - self._engine = create_engine( - self._normalize_database_url(self._settings.database_url), - future=True, - pool_pre_ping=True, - ) - - def execute_query(self, query: str, params: dict | None = None) -> list[dict]: - try: - with self._engine.connect() as connection: - result = connection.execute(text(query), params or {}) - return [self._normalize_row(dict(row._mapping)) for row in result] - except SQLAlchemyError as exc: - raise DataQueryError(f"Database query failed: {exc.__class__.__name__}: {exc}") from exc - - def _normalize_database_url(self, database_url: str) -> str: - if database_url.startswith("postgresql://"): - return database_url.replace("postgresql://", "postgresql+psycopg://", 1) - return database_url - - def _normalize_row(self, row: dict) -> dict: - normalized: dict = {} - for key, value in row.items(): - if isinstance(value, datetime): - normalized[key] = value.strftime("%d %b %Y, %I:%M:%S %p") - elif isinstance(value, date): - normalized[key] = value.strftime("%d %b %Y") - else: - normalized[key] = value - return normalized - - -data_repository = DataRepository() diff --git a/agents/analysis/backend/app/services/deterministic_analysis.py b/agents/analysis/backend/app/services/deterministic_analysis.py deleted file mode 100644 index 8f57bb1f17a461b04d904927245b22af48de7e35..0000000000000000000000000000000000000000 --- a/agents/analysis/backend/app/services/deterministic_analysis.py +++ /dev/null @@ -1,638 +0,0 @@ -from __future__ import annotations - -from decimal import Decimal, InvalidOperation -from datetime import datetime, time -from statistics import mean -from collections import defaultdict - -from agents.analysis.backend.app.models.schemas import AssetProfile - - -class DeterministicAnalysisService: - def summarize(self, asset: AssetProfile, rows: list[dict]) -> dict[str, object]: - base = self._generic_summary(rows) - domain_summary = self._asset_specific_summary(asset, rows) - base["warnings"].extend(domain_summary["warnings"]) - base["findings"] = domain_summary["findings"] - return base - - def _generic_summary(self, rows: list[dict]) -> dict[str, object]: - if not rows: - return { - "numeric_metrics": {}, - "warnings": ["No telemetry rows were returned for the requested window."], - "findings": [], - } - - numeric_columns: dict[str, list[float]] = {} - for row in rows: - for key, value in row.items(): - if isinstance(value, bool): - continue - numeric_value = self._as_float(value) - if numeric_value is not None: - numeric_columns.setdefault(key, []).append(numeric_value) - - metric_summary: dict[str, dict[str, float]] = {} - warnings: list[str] = [] - - for column, values in numeric_columns.items(): - if not values: - continue - avg = mean(values) - metric_summary[column] = { - "min": min(values), - "max": max(values), - "avg": round(avg, 3), - "latest": values[-1], - } - - if len(values) > 1 and avg != 0: - spread = max(values) - min(values) - if spread > abs(avg) * 0.2: - warnings.append( - f"{column} changed noticeably across the requested window " - f"(min={min(values):.2f}, max={max(values):.2f})." - ) - - return { - "numeric_metrics": metric_summary, - "warnings": warnings, - "findings": self._generic_trend_findings(metric_summary), - } - - def _asset_specific_summary(self, asset: AssetProfile, rows: list[dict]) -> dict[str, object]: - if asset.asset_type == "cold_room": - return self._cold_room_summary(asset, rows) - if asset.asset_type == "filling_machine": - return self._filling_machine_summary(rows) - if asset.asset_type == "tank": - return self._tank_summary(asset, rows) - if asset.asset_type == "smoke_alarm": - return self._smoke_alarm_summary(asset, rows) - return {"warnings": [], "findings": []} - - def _smoke_alarm_summary(self, asset: AssetProfile, rows: list[dict]) -> dict[str, object]: - findings: list[str] = [] - warnings: list[str] = [] - - per_asset: dict[str, list[dict]] = defaultdict(list) - for row in rows: - asset_name = str(row.get("asset_name") or asset.name or "Unknown") - per_asset[asset_name].append(row) - - if not per_asset: - return {"warnings": ["No smoke-alarm telemetry rows were available for the requested window."], "findings": []} - - smoke_events: list[tuple[str, datetime | None, str]] = [] - muted_events: list[tuple[str, datetime | None, str]] = [] - fault_events: list[tuple[str, datetime | None, str]] = [] - remove_events: list[tuple[str, datetime | None, str]] = [] - low_voltage_assets: set[str] = set() - unknown_assets: set[str] = set() - latest_states: list[str] = [] - - for asset_name, asset_rows in sorted(per_asset.items()): - ordered = sorted(asset_rows, key=lambda row: self._parse_datetime(row.get("recorded_at") or row.get("timestamp")) or datetime.min) - per_sensor: dict[str, list[dict]] = defaultdict(list) - for row in ordered: - sensor_name = str(row.get("attribute_name") or row.get("sensor_name") or "sensor") - per_sensor[sensor_name].append(row) - - for row in ordered: - warn_type = self._normalize_warn_type(row.get("warn_type")) - event_time = self._parse_datetime(row.get("recorded_at") or row.get("timestamp")) - attribute_name = str(row.get("attribute_name") or row.get("sensor_name") or "sensor") - event_ref = (asset_name, event_time, attribute_name) - if warn_type == "warn": - smoke_events.append(event_ref) - elif warn_type == "mute": - muted_events.append(event_ref) - elif warn_type == "fault": - fault_events.append(event_ref) - elif warn_type == "remove": - remove_events.append(event_ref) - elif warn_type in {"low-vol", "low-vol-test"}: - low_voltage_assets.add(asset_name) - elif warn_type == "unknown": - unknown_assets.add(asset_name) - - latest_sensor_summaries: list[str] = [] - for sensor_name, sensor_rows in sorted(per_sensor.items()): - latest_row = sensor_rows[-1] - - # Strict smoke detection: only if warn_type is 'warn' - is_actual_smoke = any(str(r.get("warn_type", "")).lower() == "warn" for r in sensor_rows) - latest_warn: str | None = "SMOKE DETECTED" if is_actual_smoke else self._normalize_warn_type(latest_row.get("warn_type")) - if is_actual_smoke: - smoke_events.append((asset_name, self._parse_datetime(latest_row.get("recorded_at") or latest_row.get("timestamp")), sensor_name)) - - latest_time = self._parse_datetime(latest_row.get("recorded_at") or latest_row.get("timestamp")) - latest_temp = self._as_float(latest_row.get("temperature")) - latest_humidity = self._as_float(latest_row.get("humidity")) - latest_battery = self._as_float(latest_row.get("bat_percent")) - - raw_data = latest_row.get("sensor_data") - if raw_data is None and latest_warn is None: - # Explicit handling for assets with no data in the window - state_fragments = [f"{sensor_name}: No recent telemetry found (Offline)"] - latest_sensor_summaries.append(" ".join(state_fragments)) - continue - - state_fragments = [f"{sensor_name}: {latest_warn or 'ok'}"] - if latest_time is not None: - state_fragments.append(f"at {latest_time.strftime('%d %b %Y, %I:%M %p')}") - - details: list[str] = [] - if latest_temp is not None: - details.append(f"{latest_temp:.1f}C") - if latest_humidity is not None: - details.append(f"{latest_humidity:.1f}% RH") - if latest_battery is not None: - details.append(f"battery {latest_battery:.0f}%") - - if isinstance(raw_data, dict): - skip_keys = {"warn", "temp", "humi", "bat_percent", "bat_voltage", "name", "timestamp"} - for k, v in raw_data.items(): - if k.lower() in skip_keys: - continue - val = self._as_float(v) - if val is not None: - details.append(f"{k}: {val}") - elif isinstance(v, str) and len(v) < 20: - details.append(f"{k}: {v}") - else: - # Strict smoke detection: only if warn_type is 'warn' - warn_type = str(latest_row.get("warn_type", "")).lower() - is_actual_smoke = (warn_type == "warn") - - if is_actual_smoke: - smoke_events.append(latest_row) - latest_warn = "SMOKE DETECTED" - # If it's a simple value, include it directly - val = self._as_float(raw_data) - if val is not None: - details.append(f"value: {val}") - elif raw_data is not None: - details.append(f"value: {raw_data}") - - if details: - state_fragments.append(f"({', '.join(details)})") - - # For 'ok' status, show sensor name and details but omit the word 'ok'. - if not latest_warn or latest_warn == 'ok': - # state_fragments[0] is like "data: ok" — keep the sensor name prefix, drop 'ok' - sensor_label = state_fragments[0].split(":")[0].strip() + ":" - ok_parts = [sensor_label] + state_fragments[1:] # sensor_name: