"""
FAM-UI - 家庭多模态智能监控系统前端 v2
深色监控面板主题 + 关键帧时间轴:
- 事件时间轴页: 左侧事件列表，右侧时间轴（时间点 + 关键帧 + 摘要）
- AI 对话 / 对话历史 / 成员命名 / 统计图表
"""
import os
import sys
import json
import base64
import html as _html
import requests
import streamlit as st
import pymysql
import pymysql.cursors
from datetime import datetime, date
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config_loader import load_config
_cfg = load_config()
_core_url = _cfg.get('core_url', 'http://127.0.0.1:8000')
_db_cfg = _cfg.get('database', {})
FRAME_DIR = _cfg.get('storage', {}).get(
'frame_image_dir', '/volume1/web/sentinel-home-ai/fam-ui/static/frames')
esc = _html.escape
def get_db_conn():
"""获取数据库连接"""
return pymysql.connect(
host=_db_cfg.get('host', '127.0.0.1'),
port=_db_cfg.get('port', 3306),
user=_db_cfg.get('user', 'root'),
password=_db_cfg.get('password', ''),
database=_db_cfg.get('database', 'sentinel_home_ai'),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
def serialize_datetime(obj):
"""序列化 datetime"""
if hasattr(obj, 'isoformat'):
return obj.isoformat()
return str(obj)
def parse_ts(ts):
"""'2026-08-14 22:31:15' / datetime -> datetime"""
if isinstance(ts, datetime):
return ts
try:
return datetime.strptime(str(ts)[:19], '%Y-%m-%d %H:%M:%S')
except ValueError:
return None
def load_frame_b64(event_id, frame_index):
"""读取落盘关键帧为 base64，不存在返回 None"""
path = os.path.join(FRAME_DIR, f'event_{event_id}', f'frame_{frame_index}.jpg')
try:
if os.path.isfile(path):
with open(path, 'rb') as f:
return base64.b64encode(f.read()).decode('ascii')
except OSError:
pass
return None
# ============================================================
# 页面配置 & 全局样式
# ============================================================
st.set_page_config(
page_title="家庭智能监控",
page_icon="🏠",
layout="wide",
initial_sidebar_state="expanded"
)
GLOBAL_CSS = """
"""
def page_header(icon: str, title: str, sub: str = ''):
st.markdown(
f''
f'
{icon}
'
f'
{esc(title)}
'
f'{f"
{esc(sub)}
" if sub else ""}'
f'
![关键帧]()
')
else:
img_html = ('📷暂无帧图
')
badges = [f'{esc(person)}']
if attention:
badges.append('⚠ 需关注')
badge_html = ''.join(badges)
action_html = esc(action) if action else '未识别到明显活动'
meta_parts = []
if clothing:
meta_parts.append(f'衣着: {esc(clothing)}')
if model:
meta_parts.append(f'模型: {esc(model)}')
meta_html = f'{" · ".join(meta_parts)}
' if meta_parts else ''
items.append(
f''
f'
{hm}{ss}
'
f'
'
f'
{img_html}'
f'
{badge_html}
'
f'
{action_html}
{meta_html}
'
f'
{"".join(items)}
',
unsafe_allow_html=True)
# ============================================================
# 侧边栏
# ============================================================
st.markdown(GLOBAL_CSS, unsafe_allow_html=True)
st.sidebar.markdown(
''
'
🏠 家庭智能监控
'
'
SENTINEL HOME AI
'
'
'
f'任务队列
'
f'待处理 {pending} · '
f'处理中 {processing} · '
f'失败 {failed}'
f'
', unsafe_allow_html=True)
except Exception:
pass
# ============================================================
# 事件时间轴页
# ============================================================
if page == "🕒 事件时间轴":
page_header('🕒', '事件时间轴', '关键时间点 · 视频帧 · 信息摘要')
with st.container():
col_date, col_sp = st.columns([1, 3])
with col_date:
date_filter = st.date_input("日期", value=None)
date_str = date_filter.isoformat() if date_filter else None
# 统计卡
try:
conn = get_db_conn()
try:
cursor = conn.cursor()
if date_str:
cursor.execute("""
SELECT COUNT(DISTINCT me.event_id) AS ev,
COUNT(ed.detail_id) AS fr,
COALESCE(SUM(ed.is_attention_event), 0) AS att,
COUNT(DISTINCT CASE WHEN ed.person <> '无人' THEN ed.person END) AS ps
FROM monitor_events me
LEFT JOIN event_details ed ON ed.event_id = me.event_id
WHERE DATE(me.event_start_time) = %s
""", (date_str,))
else:
cursor.execute("""
SELECT COUNT(DISTINCT me.event_id) AS ev,
COUNT(ed.detail_id) AS fr,
COALESCE(SUM(ed.is_attention_event), 0) AS att,
COUNT(DISTINCT CASE WHEN ed.person <> '无人' THEN ed.person END) AS ps
FROM monitor_events me
LEFT JOIN event_details ed ON ed.event_id = me.event_id
""")
s = cursor.fetchone() or {}
finally:
conn.close()
st.markdown(
f'', unsafe_allow_html=True)
except Exception:
pass
# 事件列表
page_size = 15
if 'event_page' not in st.session_state:
st.session_state.event_page = 0
offset = st.session_state.event_page * page_size
conn = get_db_conn()
try:
cursor = conn.cursor()
if date_str:
cursor.execute(
"""SELECT me.event_id, me.task_id, me.event_start_time, me.event_end_time,
me.camera_name, me.global_summary, me.compute_provider,
(SELECT COUNT(*) FROM event_details ed WHERE ed.event_id = me.event_id) AS detail_count
FROM monitor_events me
WHERE DATE(me.event_start_time) = %s
ORDER BY me.event_start_time DESC
LIMIT %s OFFSET %s""",
(date_str, page_size, offset))
else:
cursor.execute(
"""SELECT me.event_id, me.task_id, me.event_start_time, me.event_end_time,
me.camera_name, me.global_summary, me.compute_provider,
(SELECT COUNT(*) FROM event_details ed WHERE ed.event_id = me.event_id) AS detail_count
FROM monitor_events me
ORDER BY me.event_start_time DESC
LIMIT %s OFFSET %s""",
(page_size, offset))
events = cursor.fetchall()
finally:
conn.close()
if not events:
st.markdown(
'🗓'
'该日期暂无监控事件
', unsafe_allow_html=True)
else:
# 默认选中最新事件
if 'selected_event_id' not in st.session_state:
st.session_state.selected_event_id = events[0]['event_id']
# 选中的事件不在当前列表时重置
valid_ids = {e['event_id'] for e in events}
if st.session_state.selected_event_id not in valid_ids:
st.session_state.selected_event_id = events[0]['event_id']
selected_id = st.session_state.selected_event_id
col_list, col_detail = st.columns([1, 2.35], gap='large')
with col_list:
st.markdown(
f'事件列表 · {len(events)} 条
',
unsafe_allow_html=True)
for ev in events:
ts = parse_ts(ev['event_start_time'])
time_label = ts.strftime('%m-%d %H:%M') if ts else '--:--'
date_label = ts.strftime('%Y-%m-%d') if ts else ''
summary = (ev.get('global_summary') or '').strip()
if not summary:
summary = '暂无摘要'
summary_short = summary if len(summary) <= 30 else summary[:30] + '…'
is_selected = ev['event_id'] == selected_id
if st.button(
f"{'▶ ' if is_selected else ''}{time_label} · {ev.get('camera_name') or '未知'} · {ev['detail_count']}帧",
key=f"evbtn_{ev['event_id']}",
type="primary" if is_selected else "secondary",
use_container_width=True
):
st.session_state.selected_event_id = ev['event_id']
st.rerun()
date_html = (f'{date_label} · '
if date_label else '')
st.markdown(
f'{date_html}{esc(summary_short)}
',
unsafe_allow_html=True)
# 分页
nav1, nav2, nav3 = st.columns(3)
with nav1:
if st.button("← 上一页", use_container_width=True,
disabled=st.session_state.event_page == 0):
st.session_state.event_page -= 1
st.rerun()
with nav3:
if st.button("下一页 →", use_container_width=True,
disabled=len(events) < page_size):
st.session_state.event_page += 1
st.rerun()
with col_detail:
ev = next(e for e in events if e['event_id'] == selected_id)
start = parse_ts(ev['event_start_time'])
end = parse_ts(ev['event_end_time'])
range_str = ''
if start and end:
span_min = (end - start).total_seconds() / 60
range_str = (f"{start.strftime('%Y-%m-%d %H:%M:%S')} → "
f"{end.strftime('%H:%M:%S')}"
f"（约 {int(span_min)} 分钟）")
providers = ev.get('compute_provider', '[]')
if isinstance(providers, str):
try:
providers = json.loads(providers)
except ValueError:
providers = []
model_badges = ''.join(
f'{esc(p)}' for p in (providers or []))
st.markdown(
f''
f'
'
f'{esc(ev.get("camera_name") or "未知摄像头")}'
f'事件 #{ev["event_id"]}'
f'{model_badges}
'
f'
⏱ {esc(range_str)}
'
f'
{esc(ev.get("global_summary") or "暂无全局摘要")}
'
f'
🎞'
'该事件暂无关键帧明细
', unsafe_allow_html=True)
# ============================================================
# AI 对话页
# ============================================================
elif page == "💬 AI 对话":
page_header('💬', 'AI 对话', '基于监控数据的智能问答')
conn = get_db_conn()
try:
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT real_name FROM family_members WHERE real_name IS NOT NULL AND is_active = TRUE")
named = [row['real_name'] for row in cursor.fetchall()]
finally:
conn.close()
col1, col2 = st.columns(2)
with col1:
queried_person = st.text_input("查询人物", value=named[0] if named else "")
with col2:
queried_date = st.date_input("查询日期", value=date.today())
if named:
quick_person = st.selectbox("快捷选择成员", [""] + named)
if quick_person:
queried_person = quick_person
quick_questions = [
f"{queried_person}今天干嘛了？",
f"{queried_person}有没有发生什么需要注意的事情？",
f"今天{queried_person}的活动时间线是什么？",
]
selected_quick = st.selectbox("快捷提问", ["自定义"] + quick_questions)
user_question = st.text_area("你的问题", value=selected_quick if selected_quick != "自定义" else "")
if st.button("提问", type="primary"):
if not user_question.strip():
st.warning("请输入问题")
elif not queried_person.strip():
st.warning("请输入查询人物")
else:
with st.spinner("AI 正在思考..."):
try:
resp = requests.post(
f"{_core_url}/api/chat/ask",
json={
"question": user_question,
"queried_person": queried_person,
"queried_date": queried_date.isoformat()
},
timeout=120
)
if resp.status_code == 200:
data = resp.json()
st.markdown(
f'❓ 提问
'
f'{esc(user_question)}
💬'
'暂无对话记录
', unsafe_allow_html=True)
else:
for h in history:
created = serialize_datetime(h['created_at'])
st.markdown(
f''
f'👤 {esc(h.get("queried_person") or "未知")} · {esc(created)}
'
f'{esc(h["user_question"])}
第 {st.session_state.chat_page + 1} 页
',
unsafe_allow_html=True)
with nav3:
if st.button("下一页 →", disabled=len(history) < page_size):
st.session_state.chat_page += 1
st.rerun()
# ============================================================
# 成员命名页
# ============================================================
elif page == "👤 成员命名":
page_header('👤', '家庭成员命名', '为识别到的人物起名')
conn = get_db_conn()
try:
cursor = conn.cursor()
st.markdown('未命名人物
', unsafe_allow_html=True)
cursor.execute(
"""SELECT fm.abstract_label, fm.feature_description, fm.first_seen_at,
(SELECT COUNT(*) FROM event_details ed WHERE ed.person = fm.abstract_label) AS event_count
FROM family_members fm
WHERE fm.real_name IS NULL AND fm.is_active = TRUE
ORDER BY fm.first_seen_at ASC"""
)
unnamed = cursor.fetchall()
if not unnamed:
st.markdown(
'✅'
'所有人物已命名，或暂未发现新人物
', unsafe_allow_html=True)
else:
for m in unnamed:
col1, col2, col3 = st.columns([2, 1.6, 1])
with col1:
first_seen = serialize_datetime(m['first_seen_at'])
st.markdown(
f''
f'
'
f'{esc(m["abstract_label"])}
'
f'
{esc(m["feature_description"] or "无特征描述")}
'
f'
'
f'首次出现 {esc(first_seen)} · 事件数 {m["event_count"]}
'
f'
已命名成员
', unsafe_allow_html=True)
cursor.execute(
"""SELECT abstract_label, real_name, feature_description, first_seen_at, named_at, named_by
FROM family_members
WHERE real_name IS NOT NULL AND is_active = TRUE
ORDER BY named_at DESC"""
)
named = cursor.fetchall()
if not named:
st.markdown(
'暂无已命名成员
', unsafe_allow_html=True)
else:
cards = []
for m in named:
cards.append(
f''
f'
'
f'{esc(m["real_name"])} '
f''
f'{esc(m["abstract_label"])}
'
f'
'
f'{esc(m["feature_description"] or "")}
'
f'
'
f'首次出现 {esc(serialize_datetime(m["first_seen_at"]))} · '
f'命名于 {esc(serialize_datetime(m["named_at"]))}
'
f'
{"".join(cards)}
',
unsafe_allow_html=True)
finally:
conn.close()
# ============================================================
# 统计图表页
# ============================================================
elif page == "📈 统计图表":
page_header('📈', '统计图表', '模型来源 / 关注事件 / 任务状态')
conn = get_db_conn()
try:
cursor = conn.cursor()
st.markdown('模型来源分布
', unsafe_allow_html=True)
try:
cursor.execute("""
SELECT JSON_UNQUOTE(JSON_EXTRACT(item, '$')) AS provider, COUNT(*) AS count
FROM monitor_events,
JSON_TABLE(compute_provider, '$[*]'
COLUMNS(item VARCHAR(50) PATH '$')
) AS jt
GROUP BY provider ORDER BY count DESC
""")
stats = cursor.fetchall()
except Exception:
cursor.execute("SELECT compute_provider, COUNT(*) AS count FROM monitor_events GROUP BY compute_provider")
stats = cursor.fetchall()
if stats:
st.bar_chart({r['provider']: r['count'] for r in stats})
else:
st.markdown('暂无统计数据
',
unsafe_allow_html=True)
st.markdown('关注事件统计
', unsafe_allow_html=True)
cursor.execute("""
SELECT DATE(frame_timestamp) AS date, person, COUNT(*) AS count
FROM event_details
WHERE is_attention_event = TRUE
GROUP BY DATE(frame_timestamp), person
ORDER BY date DESC
""")
attention = cursor.fetchall()
if attention:
import pandas as pd
df_att = pd.DataFrame(attention)
st.dataframe(df_att, use_container_width=True, hide_index=True)
else:
st.markdown('暂无关注事件
',
unsafe_allow_html=True)
st.markdown('任务状态统计
', unsafe_allow_html=True)
cursor.execute("""
SELECT status, COUNT(*) AS count
FROM process_tasks
GROUP BY status
""")
task_stats = cursor.fetchall()
if task_stats:
st.bar_chart({r['status']: r['count'] for r in task_stats})
else:
st.markdown('暂无任务数据
',
unsafe_allow_html=True)
finally:
conn.close()