#!/usr/bin/env python3
"""
System hardware and services → Brain index pipeline.
Captures a snapshot of system specs, running services, and network ports.
"""
import json
import subprocess
import sys
from pathlib import Path
BRAIN_DIR = Path.home() / '.hermes' / 'brain'
sys.path.insert(0, str(BRAIN_DIR))
from brain import init_brain, log_event, query_events
def run(cmd, default=''):
"""Run a shell command safely."""
try:
return subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=10
).stdout.strip() or default
except Exception:
return default
def purge_existing():
"""Remove previously ingested system events."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='system_index', limit=10000)
if not events:
print(" No existing system events to purge")
return 0
brain_dir = Path.home() / '.hermes' / 'brain'
db_path = brain_dir / 'index' / 'events.db'
event_ids = [e['id'] for e in events]
try:
import sqlite3
db = sqlite3.connect(str(db_path))
batch_size = 500
for i in range(0, len(event_ids), batch_size):
batch = event_ids[i:i+batch_size]
placeholders = ','.join(['?'] * len(batch))
db.execute(f"DELETE FROM events WHERE id IN ({placeholders})", batch)
db.commit()
db.close()
except Exception as e:
print(f" Warning: purge error: {e}")
return len(event_ids)
def ingest_system():
"""Capture and ingest system snapshot."""
init_brain()
print("Purging existing system events...")
purged = purge_existing()
if purged:
print(f" Purged {purged} events")
print("\nCapturing system snapshot...")
count = 0
# 1. Hardware inventory
cpu = run('lscpu | grep "Model name" | cut -d: -f2 | xargs')
cpu_cores = run('lscpu | grep "^CPU(s):" | head -1 | cut -d: -f2 | xargs')
gpu = run('nvidia-smi --query-gpu=name,memory.total --format=csv,noheader 2>/dev/null')
gpu = gpu.strip() if gpu else 'None detected'
ram = run('free -h | grep Mem | awk \'{print $2}\'}')
ram_used = run('free -h | grep Mem | awk \'{print $3}\'}')
disks = run('lsblk -d -o NAME,SIZE,TYPE,MOUNTPOINT 2>/dev/null | grep disk')
hostname = run('hostname')
os_info = run('cat /etc/os-release | grep PRETTY_NAME | cut -d= -f2 | tr -d \\"')
kernel = run('uname -r')
tags = ['system', 'hardware', 'inventory', 'server']
context = {
'type': 'hardware',
'hostname': hostname,
'os': os_info,
'kernel': kernel,
'cpu': cpu,
'cpu_cores': cpu_cores,
'gpu': gpu,
'ram': ram,
'ram_used': ram_used,
'disks': disks,
}
log_event(
event_type='knowledge_ingest',
tool='system_index',
args={'type': 'hardware'},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f" [{count}] Hardware: {cpu}")
print(f" GPU: {gpu}")
print(f" RAM: {ram} ({ram_used} used)")
print(f" OS: {os_info} {kernel}")
# 2. Running services
services_raw = run('systemctl list-units --type=service --state=running --no-pager --no-legend')
services = []
for line in services_raw.split('\n'):
parts = line.split()
if parts:
svc_name = parts[0].replace('.service', '')
services.append(svc_name)
tags = ['system', 'services', 'running']
context = {
'type': 'services',
'count': len(services),
'services': services,
}
log_event(
event_type='knowledge_ingest',
tool='system_index',
args={'type': 'services'},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f"\n [{count}] Services: {len(services)} running")
# 3. Network ports
ports_raw = run('ss -tlnp')
ports = []
for line in ports_raw.split('\n')[1:]:
parts = line.split()
if len(parts) >= 4:
local = parts[3]
addr, port = local.rsplit(':', 1)
process = ''
if len(parts) >= 5:
process = parts[4].split(')')[0].split('(')[-1]
ports.append({'address': addr, 'port': port, 'process': process})
tags = ['system', 'network', 'ports']
context = {
'type': 'network',
'ports': ports,
}
log_event(
event_type='knowledge_ingest',
tool='system_index',
args={'type': 'network'},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f" [{count}] Network ports: {len(ports)} listeners")
for p in ports:
print(f" {p['address']}:{p['port']} ({p['process']})")
# 4. Docker containers
docker = run('docker ps --format "{{.Names}}\\t{{.Status}}\\t{{.Ports}}"')
containers = []
if docker:
for line in docker.split('\n'):
if line.strip():
parts = line.split('\t')
containers.append({
'name': parts[0] if len(parts) > 0 else '',
'status': parts[1] if len(parts) > 1 else '',
'ports': parts[2] if len(parts) > 2 else '',
})
tags = ['system', 'docker', 'containers']
context = {
'type': 'docker',
'count': len(containers),
'containers': containers,
}
log_event(
event_type='knowledge_ingest',
tool='system_index',
args={'type': 'docker'},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f"\n [{count}] Docker containers: {len(containers)}")
return count
def test_queries():
"""Verify system index with queries."""
init_brain()
queries = [
("hardware specs", "Should find hardware info"),
("AMD Ryzen", "Should find CPU"),
("RTX 4090", "Should find GPU"),
("running services", "Should find service list"),
("docker container", "Should find Docker info"),
("port 8081", "Should find freqtrade port"),
("network", "Should find network info"),
]
print("\n=== TEST QUERIES ===")
for query, expected in queries:
results = query_events(
event_type='knowledge_ingest',
tool='system_index',
full_text_search=query,
limit=2,
)
status = "✓" if results else "✗"
print(f" {status} '{query}' → {len(results)} results ({expected})")
if __name__ == '__main__':
print("=" * 60)
print("System Hardware & Services → Brain Index Pipeline")
print("=" * 60)
count = ingest_system()
print(f"\nTotal ingested: {count}")
test_queries()
print("\nDone.")