350 lines
13 KiB
Python
350 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
RAG Log Ingester
|
|
Feeds Frigate events, HA logs, HA history and Docker logs into ChromaDB
|
|
using nomic-embed-text embeddings from Ollama.
|
|
Runs on a schedule (default every 5 min).
|
|
"""
|
|
|
|
import os, json, time, hashlib, sqlite3, datetime, logging, requests, schedule
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s"
|
|
)
|
|
log = logging.getLogger("ingester")
|
|
|
|
# ── Config ────────────────────────────────────────────────────────────────────
|
|
CHROMA_HOST = os.getenv("CHROMA_HOST", "chromadb")
|
|
CHROMA_PORT = int(os.getenv("CHROMA_PORT", "8000"))
|
|
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://ollama:11434")
|
|
FRIGATE_DB = os.getenv("FRIGATE_DB", "/data/frigate/frigate.db")
|
|
HA_LOG = os.getenv("HA_LOG", "/data/ha/home-assistant.log")
|
|
STATE_FILE = os.getenv("STATE_FILE", "/data/state/state.json")
|
|
INTERVAL = int(os.getenv("INGEST_INTERVAL", "300"))
|
|
EMBED_MODEL = "nomic-embed-text"
|
|
|
|
# PostgreSQL (HA recorder)
|
|
PG_HOST = os.getenv("PG_HOST", "postgres18")
|
|
PG_PORT = int(os.getenv("PG_PORT", "5432"))
|
|
PG_DB = os.getenv("PG_DB", "homeassistant")
|
|
PG_USER = os.getenv("PG_USER", "homeassistant")
|
|
PG_PASS = os.getenv("PG_PASS", "")
|
|
|
|
DOCKER_CONTAINERS = [
|
|
"homeassistant", "frigate", "mosquitto", "esphome",
|
|
"traefik", "crowdsec", "immich_server", "ha-dashboard"
|
|
]
|
|
|
|
# ── State persistence ─────────────────────────────────────────────────────────
|
|
def load_state():
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
def save_state(state):
|
|
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(state, f, indent=2, default=str)
|
|
|
|
# ── Embeddings via Ollama ─────────────────────────────────────────────────────
|
|
def embed_texts(texts: list[str]) -> list[list[float] | None]:
|
|
result = []
|
|
for text in texts:
|
|
try:
|
|
r = requests.post(
|
|
f"{OLLAMA_URL}/api/embeddings",
|
|
json={"model": EMBED_MODEL, "prompt": text[:4000]},
|
|
timeout=60,
|
|
)
|
|
r.raise_for_status()
|
|
result.append(r.json()["embedding"])
|
|
except Exception as e:
|
|
log.warning(f"Embedding failed: {e}")
|
|
result.append(None)
|
|
return result
|
|
|
|
def doc_id(text: str) -> str:
|
|
return hashlib.sha256(text.encode()).hexdigest()[:32]
|
|
|
|
# ── ChromaDB helpers ──────────────────────────────────────────────────────────
|
|
def chroma_client():
|
|
import chromadb
|
|
return chromadb.HttpClient(host=CHROMA_HOST, port=CHROMA_PORT)
|
|
|
|
def get_collections(client):
|
|
return {
|
|
"frigate": client.get_or_create_collection("frigate_events"),
|
|
"ha_log": client.get_or_create_collection("ha_logs"),
|
|
"ha_history": client.get_or_create_collection("ha_history"),
|
|
"docker": client.get_or_create_collection("docker_logs"),
|
|
}
|
|
|
|
def upsert_docs(collection, docs: list[str], metadatas: list[dict], ids: list[str] | None = None):
|
|
if not docs:
|
|
return
|
|
if ids is None:
|
|
ids = [doc_id(d) for d in docs]
|
|
# Deduplicate within batch (keep last occurrence)
|
|
seen = {}
|
|
for i, id_ in enumerate(ids):
|
|
seen[id_] = i
|
|
keep = sorted(seen.values())
|
|
docs = [docs[i] for i in keep]
|
|
metadatas = [metadatas[i] for i in keep]
|
|
ids = [ids[i] for i in keep]
|
|
embeddings = embed_texts(docs)
|
|
|
|
# Filter out failed embeddings
|
|
valid = [(i, d, m, e) for i, d, m, e in zip(ids, docs, metadatas, embeddings) if e]
|
|
if not valid:
|
|
return
|
|
ids_v, docs_v, meta_v, emb_v = zip(*valid)
|
|
|
|
try:
|
|
collection.upsert(
|
|
ids=list(ids_v),
|
|
documents=list(docs_v),
|
|
metadatas=list(meta_v),
|
|
embeddings=list(emb_v),
|
|
)
|
|
log.info(f"Upserted {len(ids_v)} docs → {collection.name}")
|
|
except Exception as e:
|
|
log.error(f"ChromaDB upsert failed ({collection.name}): {e}")
|
|
|
|
# ── Source: Frigate SQLite ────────────────────────────────────────────────────
|
|
def ingest_frigate(col, state: dict) -> dict:
|
|
if not os.path.exists(FRIGATE_DB):
|
|
log.warning(f"Frigate DB not found: {FRIGATE_DB}")
|
|
return state
|
|
|
|
last_id = state.get("frigate_last_id", "")
|
|
try:
|
|
conn = sqlite3.connect(f"file:{FRIGATE_DB}?immutable=1", uri=True)
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT id, label, camera, start_time, end_time, top_score AS score,
|
|
zones, false_positive, has_snapshot, has_clip
|
|
FROM event
|
|
WHERE id > ? AND false_positive = 0
|
|
ORDER BY start_time ASC
|
|
LIMIT 500
|
|
""", (last_id,))
|
|
rows = cur.fetchall()
|
|
conn.close()
|
|
except Exception as e:
|
|
log.error(f"Frigate DB error: {e}")
|
|
return state
|
|
|
|
docs, metas, new_last_id = [], [], last_id
|
|
for row in rows:
|
|
start = datetime.datetime.fromtimestamp(row["start_time"]).strftime("%Y-%m-%d %H:%M:%S")
|
|
duration = f"{int(row['end_time'] - row['start_time'])}s" if row["end_time"] else "ongoing"
|
|
try:
|
|
zones = json.loads(row["zones"] or "[]")
|
|
except Exception:
|
|
zones = []
|
|
zones_str = f", zones: {', '.join(zones)}" if zones else ""
|
|
score = f"{int((row['score'] or 0) * 100)}%"
|
|
|
|
text = (
|
|
f"Frigate: {row['label']} detected on camera {row['camera']} "
|
|
f"at {start}, duration {duration}, confidence {score}{zones_str}."
|
|
)
|
|
docs.append(text)
|
|
metas.append({
|
|
"source": "frigate",
|
|
"camera": str(row["camera"]),
|
|
"label": str(row["label"]),
|
|
"timestamp": start,
|
|
"score": float(row["score"] or 0),
|
|
"has_clip": int(row["has_clip"] or 0),
|
|
})
|
|
new_last_id = row["id"]
|
|
|
|
upsert_docs(col, docs, metas)
|
|
state["frigate_last_id"] = new_last_id
|
|
return state
|
|
|
|
# ── Source: HA recorder (PostgreSQL) ─────────────────────────────────────────
|
|
def ingest_ha_history(col, state: dict) -> dict:
|
|
if not PG_PASS:
|
|
log.warning("PG_PASS not set, skipping HA history")
|
|
return state
|
|
|
|
# Default: start 7 days back (not from epoch — HA has 842K+ records)
|
|
seven_days_ago = (datetime.datetime.utcnow() - datetime.timedelta(days=7)).timestamp()
|
|
last_ts = state.get("ha_history_last_ts", seven_days_ago)
|
|
try:
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
conn = psycopg2.connect(
|
|
host=PG_HOST, port=PG_PORT, dbname=PG_DB,
|
|
user=PG_USER, password=PG_PASS,
|
|
connect_timeout=10,
|
|
)
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
|
cur.execute("""
|
|
SELECT sm.entity_id, s.state, s.last_changed_ts, sa.shared_attrs
|
|
FROM states s
|
|
JOIN states_meta sm ON s.metadata_id = sm.metadata_id
|
|
LEFT JOIN state_attributes sa ON s.attributes_id = sa.attributes_id
|
|
WHERE s.last_changed_ts > %s
|
|
AND s.state NOT IN ('unavailable', 'unknown', '')
|
|
AND sm.entity_id NOT LIKE 'camera.%%'
|
|
AND sm.entity_id NOT LIKE 'media_player.%%'
|
|
AND sm.entity_id NOT LIKE 'update.%%'
|
|
ORDER BY s.last_changed_ts ASC
|
|
LIMIT 500
|
|
""", (last_ts,))
|
|
rows = cur.fetchall()
|
|
conn.close()
|
|
except Exception as e:
|
|
log.error(f"HA PostgreSQL error: {e}")
|
|
return state
|
|
|
|
CHUNK = 50
|
|
docs, metas, ids, new_last_ts = [], [], [], last_ts
|
|
for row in rows:
|
|
try:
|
|
attrs = json.loads(row["shared_attrs"] or "{}")
|
|
except Exception:
|
|
attrs = {}
|
|
friendly = attrs.get("friendly_name", row["entity_id"])
|
|
unit = attrs.get("unit_of_measurement", "")
|
|
ts_raw = row["last_changed_ts"]
|
|
ts = datetime.datetime.fromtimestamp(float(ts_raw)).strftime("%Y-%m-%d %H:%M:%S") if ts_raw else ""
|
|
state_val = row["state"]
|
|
|
|
text = (
|
|
f"HA: {friendly} ({row['entity_id']}) = "
|
|
f"'{state_val}{(' ' + unit) if unit else ''}' at {ts} (ts:{ts_raw})."
|
|
)
|
|
docs.append(text)
|
|
metas.append({
|
|
"source": "ha_history",
|
|
"entity_id": str(row["entity_id"]),
|
|
"state": str(state_val),
|
|
"timestamp": ts,
|
|
})
|
|
ids.append(doc_id(f"{row['entity_id']}|{ts_raw}|{state_val}"))
|
|
new_last_ts = max(new_last_ts, float(ts_raw or 0))
|
|
|
|
# Upsert in chunks to avoid payload size issues
|
|
if len(docs) >= CHUNK:
|
|
upsert_docs(col, docs, metas, ids)
|
|
docs, metas, ids = [], [], []
|
|
|
|
if docs:
|
|
upsert_docs(col, docs, metas, ids)
|
|
|
|
state["ha_history_last_ts"] = new_last_ts
|
|
log.info(f"HA history: processed {len(rows)} records up to ts={new_last_ts}")
|
|
return state
|
|
|
|
# ── Source: HA log file ───────────────────────────────────────────────────────
|
|
def ingest_ha_log(col, state: dict) -> dict:
|
|
if not os.path.exists(HA_LOG):
|
|
log.warning(f"HA log not found: {HA_LOG}")
|
|
return state
|
|
|
|
last_pos = state.get("ha_log_pos", 0)
|
|
try:
|
|
with open(HA_LOG, "r", errors="replace") as f:
|
|
f.seek(last_pos)
|
|
lines = f.readlines()
|
|
new_pos = f.tell()
|
|
except Exception as e:
|
|
log.error(f"HA log error: {e}")
|
|
return state
|
|
|
|
chunk_size = 25
|
|
docs, metas = [], []
|
|
for i in range(0, len(lines), chunk_size):
|
|
chunk = "".join(lines[i:i + chunk_size]).strip()
|
|
if not chunk:
|
|
continue
|
|
ts = lines[i][:23] if lines[i] else ""
|
|
docs.append(chunk)
|
|
metas.append({"source": "ha_log", "timestamp": ts})
|
|
|
|
upsert_docs(col, docs, metas)
|
|
state["ha_log_pos"] = new_pos
|
|
return state
|
|
|
|
# ── Source: Docker logs ───────────────────────────────────────────────────────
|
|
def ingest_docker_logs(col, state: dict) -> dict:
|
|
try:
|
|
import docker as dockerlib
|
|
dclient = dockerlib.from_env()
|
|
except Exception as e:
|
|
log.error(f"Docker error: {e}")
|
|
return state
|
|
|
|
containers = dclient.containers.list()
|
|
if DOCKER_CONTAINERS:
|
|
containers = [c for c in containers if c.name in DOCKER_CONTAINERS]
|
|
|
|
for container in containers:
|
|
key = f"docker_since_{container.name}"
|
|
since_ts = state.get(key)
|
|
try:
|
|
kwargs = {"timestamps": True}
|
|
if since_ts:
|
|
kwargs["since"] = int(since_ts)
|
|
else:
|
|
kwargs["tail"] = 300
|
|
|
|
raw = container.logs(**kwargs).decode("utf-8", errors="replace")
|
|
lines = [l for l in raw.splitlines() if l.strip()]
|
|
if not lines:
|
|
continue
|
|
|
|
chunk_size = 30
|
|
docs, metas = [], []
|
|
for i in range(0, len(lines), chunk_size):
|
|
chunk = "\n".join(lines[i:i + chunk_size])
|
|
ts = lines[i][:30]
|
|
docs.append(f"[{container.name}]\n{chunk}")
|
|
metas.append({"source": "docker", "container": container.name, "timestamp": ts})
|
|
|
|
upsert_docs(col, docs, metas)
|
|
state[key] = int(time.time())
|
|
|
|
except Exception as e:
|
|
log.warning(f"Docker log error ({container.name}): {e}")
|
|
|
|
return state
|
|
|
|
# ── Main loop ─────────────────────────────────────────────────────────────────
|
|
def run_ingestion():
|
|
log.info("=== Ingestion cycle start ===")
|
|
try:
|
|
client = chroma_client()
|
|
cols = get_collections(client)
|
|
except Exception as e:
|
|
log.error(f"ChromaDB connection failed: {e}")
|
|
return
|
|
|
|
state = load_state()
|
|
state = ingest_frigate(cols["frigate"], state)
|
|
state = ingest_ha_history(cols["ha_history"], state)
|
|
state = ingest_ha_log(cols["ha_log"], state)
|
|
state = ingest_docker_logs(cols["docker"], state)
|
|
save_state(state)
|
|
log.info("=== Ingestion cycle done ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
log.info(f"Log ingester starting — interval={INTERVAL}s")
|
|
# Wait for dependencies
|
|
time.sleep(15)
|
|
run_ingestion()
|
|
schedule.every(INTERVAL).seconds.do(run_ingestion)
|
|
while True:
|
|
schedule.run_pending()
|
|
time.sleep(10)
|