mao1/shenyan-app/cyberboss/pare.py

#!/usr/bin/env python3
"""pare — 身体数据陡度监控。检测到异常波动时通知沈晏判断是否推消息。"""
import json, os, time, sys

INBOX = os.path.expanduser("~/.shenyan-bridge/inbox.jsonl")
STATE_FILE = os.path.expanduser("~/.cyberboss/pare_state.json")
LOG = os.path.expanduser(f"~/.cyberboss/logs/pare-{time.strftime('%Y-%m-%d')}.log")
BRIDGE_NOTES = "http://localhost:3004/notes/add"

# thresholds — steepness triggers
HR_SPIKE = 20       # bpm jump from last reading
HR_HIGH = 120       # absolute high
HR_LOW = 50         # absolute low
STEP_BURST = 2000   # sudden steps in 10 min
STEP_LOW = 100      # very low activity alert (half day)
SLEEP_SHORT = 4     # hours — short sleep alert

def load_state():
    if os.path.exists(STATE_FILE):
        try: return json.load(open(STATE_FILE))
        except: pass
    return {"heart_rates": [], "steps_history": [], "last_alert": 0}

def save_state(state):
    # keep only last 20 readings for trend
    state["heart_rates"] = state.get("heart_rates", [])[-20:]
    state["steps_history"] = state.get("steps_history", [])[-20:]
    json.dump(state, open(STATE_FILE, "w"))

def read_latest_health():
    """parse latest [health] entry from inbox"""
    if not os.path.exists(INBOX): return None
    try:
        lines = open(INBOX).read().strip().split("\n")
        for line in reversed(lines):
            entry = json.loads(line)
            if isinstance(entry.get("text"), str) and entry["text"].startswith("[health]"):
                payload = entry["text"][8:]  # remove "[health] " prefix
                return json.loads(payload)
    except: pass
    return None

def check_heart_rate(hr, state):
    """check heart rate for spikes or dangerous levels"""
    alerts = []
    history = state.get("heart_rates", [])

    if hr is None: return alerts
    bpm = hr.get("bpm", 0)
    if bpm == 0: return alerts

    # spike from last reading
    if history:
        last = history[-1].get("bpm", 0)
        delta = abs(bpm - last)
        if delta >= HR_SPIKE:
            direction = "升高" if bpm > last else "降低"
            alerts.append(f"心率{direction} {delta}bpm: {last}→{bpm}")

    # absolute thresholds
    if bpm >= HR_HIGH:
        alerts.append(f"心率过高: {bpm}bpm")
    elif bpm <= HR_LOW:
        alerts.append(f"心率过低: {bpm}bpm")

    history.append(hr)
    return alerts

def check_steps(steps, state):
    """check step activity patterns"""
    alerts = []
    history = state.get("steps_history", [])

    if steps is None: return alerts
    if steps == 0: return alerts

    history.append({"steps": steps, "time": time.time()})

    # very low activity — only alert once per 6h
    now = time.time()
    if steps < STEP_LOW and (now - state.get("last_alert", 0)) > 21600:
        alerts.append(f"活动量偏低: {steps}步")

    return alerts

def check_sleep(sleep, state):
    """check sleep quality"""
    alerts = []
    if sleep is None: return alerts
    hours = sleep.get("hours", 0)
    if hours > 0 and hours < SLEEP_SHORT:
        alerts.append(f"睡眠不足: {hours:.1f}小时")
    return alerts

def log(msg):
    timestamp = time.strftime("%H:%M:%S")
    line = f"[{timestamp}] {msg}"
    print(line)
    with open(LOG, "a") as f:
        f.write(line + "\n")

def push_alert(text):
    """write alert to bridge inbox + fridge notes"""
    entry = {
        "id": int(time.time() * 1000) % 100000,
        "type": "pare",
        "text": f"[pare] {text}",
        "time": int(time.time() * 1000),
        "status": "pending"
    }
    with open(INBOX, "a") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    # also post to fridge notes
    try:
        import urllib.request
        req = urllib.request.Request(
            BRIDGE_NOTES,
            data=json.dumps({"title": "pare", "body": text}).encode(),
            headers={"Content-Type": "application/json"}
        )
        urllib.request.urlopen(req, timeout=5)
    except: pass

def main():
    state = load_state()
    health = read_latest_health()

    if health is None:
        return  # no data yet, silent

    all_alerts = []
    all_alerts += check_heart_rate(health.get("heartRate"), state)
    all_alerts += check_steps(health.get("steps"), state)
    all_alerts += check_sleep(health.get("sleep"), state)

    if all_alerts:
        for a in all_alerts:
            log(f"ALERT: {a}")
            push_alert(a)
        state["last_alert"] = time.time()
    else:
        # log normal status once per hour
        hr = health.get("heartRate", {}).get("bpm", "?")
        steps = health.get("steps", "?")
        log(f"normal — HR:{hr} Steps:{steps}")

    save_state(state)

if __name__ == "__main__":
    main()