feat: sync spo2 stress and abnormal heart beat

This commit is contained in:
binglua committed 2026-06-22 01:31:54 +00:00
1 parent b532ee652d
commit 0ff3dd1c82
8 files changed
+596 -3

No files matched your search

+30
View File
@@ -5,10 +5,13 @@ from collections.abc import AsyncIterator, Coroutine, Iterator
from typing import Any, Union
from mi_fitness_mcp.models import (
AbnormalHeartBeatEvent,
BodyMeasurement,
DailyActivity,
HeartRateSample,
SleepSession,
SpO2Sample,
StressSample,
Workout,
)
@@ -80,6 +83,33 @@ class DataAdapter(ABC):
"""Iterate over heart rate records."""
pass
@abstractmethod
def iter_spo2(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> Iterator[SpO2Sample] | AsyncIterator[SpO2Sample]:
"""Iterate over blood oxygen saturation records."""
pass
@abstractmethod
def iter_stress(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> Iterator[StressSample] | AsyncIterator[StressSample]:
"""Iterate over stress records."""
pass
@abstractmethod
def iter_abnormal_heart_beat(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> Iterator[AbnormalHeartBeatEvent] | AsyncIterator[AbnormalHeartBeatEvent]:
"""Iterate over abnormal heart beat events."""
pass
@abstractmethod
def get_available_data_types(self) -> list[str]:
"""Get list of available data types."""
@@ -14,11 +14,14 @@ import httpx
from mi_fitness_mcp.adapters.base import DataAdapter
from mi_fitness_mcp.models import (
AbnormalHeartBeatEvent,
BodyMeasurement,
DailyActivity,
HeartRateSample,
SleepSession,
SleepStage,
SpO2Sample,
StressSample,
Workout,
)
@@ -235,6 +238,9 @@ class MiFitnessCloudAdapter(DataAdapter):
"heart_rate": "heart_rate",
"body_measurements": "weight",
"sleep": "sleep",
"spo2": "spo2",
"stress": "stress",
"abnormal_heart_beat": "abnormal_heart_beat",
}
for data_type, key in probes.items():
try:
@@ -537,12 +543,133 @@ class MiFitnessCloudAdapter(DataAdapter):
id=f"mi_fitness_hr_{int(item.get('time', 0))}",
provider="mi_fitness",
source_type="cloud_session",
source_record_id=str(item.get("time", "")) or None,
user_id=self.user_id or "unknown",
timezone=item.get("zone_name") or "UTC",
collected_at=self._record_datetime(item),
timestamp=self._record_datetime(item),
bpm=int(payload.get("bpm", 0)),
sample_type=sample_type,
)
resting_records = await self._fetch_key("resting_heart_rate", start_date, end_date)
for item in resting_records:
payload = self._parse_value(item)
timestamp = payload.get("date_time") or item.get("time")
yield HeartRateSample(
id=f"mi_fitness_resting_hr_{int(timestamp or item.get('time', 0))}",
provider="mi_fitness",
source_type="cloud_session",
source_record_id=str(item.get("time", "")) or None,
user_id=self.user_id or "unknown",
timezone=item.get("zone_name") or "UTC",
collected_at=self._record_datetime(item),
timestamp=self._timestamp_to_datetime(timestamp or item.get("time"), int(item.get("zone_offset", 0) or 0)),
bpm=int(payload.get("bpm", 0)),
sample_type="resting",
)
def _stress_level(self, score: int) -> str:
if score < 30:
return "low"
if score < 60:
return "medium"
return "high"
async def iter_spo2(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> AsyncIterator[SpO2Sample]:
if not self.is_connected() or not start_date or not end_date:
return
yield
records = await self._fetch_key("spo2", start_date, end_date)
for item in records:
payload = self._parse_value(item)
timestamp = payload.get("time") or item.get("time")
spo2 = payload.get("spo2") or payload.get("value")
if timestamp is None or spo2 is None:
continue
yield SpO2Sample(
id=f"mi_fitness_spo2_{int(timestamp)}",
provider="mi_fitness",
source_type="cloud_session",
source_record_id=str(item.get("time", "")) or None,
user_id=self.user_id or "unknown",
timezone=item.get("zone_name") or "UTC",
collected_at=self._record_datetime(item),
timestamp=self._timestamp_to_datetime(timestamp, int(item.get("zone_offset", 0) or 0)),
spo2_pct=int(float(spo2)),
)
async def iter_stress(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> AsyncIterator[StressSample]:
if not self.is_connected() or not start_date or not end_date:
return
yield
records = await self._fetch_key("stress", start_date, end_date)
for item in records:
payload = self._parse_value(item)
timestamp = payload.get("time") or item.get("time")
stress = payload.get("stress") or payload.get("score") or payload.get("value")
if timestamp is None or stress is None:
continue
score = int(float(stress))
yield StressSample(
id=f"mi_fitness_stress_{int(timestamp)}",
provider="mi_fitness",
source_type="cloud_session",
source_record_id=str(item.get("time", "")) or None,
user_id=self.user_id or "unknown",
timezone=item.get("zone_name") or "UTC",
collected_at=self._record_datetime(item),
timestamp=self._timestamp_to_datetime(timestamp, int(item.get("zone_offset", 0) or 0)),
stress_score=score,
level=self._stress_level(score),
)
async def iter_abnormal_heart_beat(
self,
start_date: str | None = None,
end_date: str | None = None,
) -> AsyncIterator[AbnormalHeartBeatEvent]:
if not self.is_connected() or not start_date or not end_date:
return
yield
records = await self._fetch_key("abnormal_heart_beat", start_date, end_date)
for item in records:
payload = self._parse_value(item)
zone_offset = int(item.get("zone_offset", 0) or 0)
start_ts = payload.get("start_time") or item.get("time")
end_ts = payload.get("end_time") or start_ts
if start_ts is None:
continue
start_at = self._timestamp_to_datetime(start_ts, zone_offset)
end_at = self._timestamp_to_datetime(end_ts, zone_offset)
duration_seconds = max(0, int(end_ts) - int(start_ts))
event_id = f"{item.get('sid', self.user_id)}_{int(start_ts)}"
yield AbnormalHeartBeatEvent(
id=f"mi_fitness_abnormal_hr_{event_id}",
provider="mi_fitness",
source_type="cloud_session",
source_record_id=str(item.get("time", "")) or None,
user_id=self.user_id or "unknown",
timezone=item.get("zone_name") or "UTC",
collected_at=self._record_datetime(item),
event_id=event_id,
start_at=start_at,
end_at=end_at,
duration_seconds=duration_seconds,
)
async def close(self) -> None:
await self._close_client()
self._connected = False
+10 -1
View File
@@ -162,7 +162,16 @@ def main():
sync_parser = subparsers.add_parser("sync", help="Sync data from source")
sync_parser.add_argument(
"--type",
choices=["daily_activity", "body_measurements", "heart_rate", "sleep", "workouts"],
choices=[
"daily_activity",
"body_measurements",
"heart_rate",
"sleep",
"workouts",
"spo2",
"stress",
"abnormal_heart_beat",
],
help="Type of data to sync",
)
sync_parser.add_argument("--start-date", help="Start date (YYYY-MM-DD)")
+10 -1
View File
@@ -102,7 +102,6 @@ class SpO2Sample(BaseEntity):
timestamp: datetime = Field(description="Measurement time")
spo2_pct: int = Field(ge=0, le=100, description="SpO2 percentage")
class StressSample(BaseEntity):
"""Stress level measurement."""
@@ -111,6 +110,16 @@ class StressSample(BaseEntity):
level: Literal["low", "medium", "high"] = Field(description="Stress level category")
class AbnormalHeartBeatEvent(BaseEntity):
"""Abnormal heart beat event."""
event_id: str = Field(description="Unique abnormal heart beat event ID")
start_at: datetime = Field(description="Event start time")
end_at: datetime = Field(description="Event end time")
duration_seconds: int = Field(ge=0, description="Event duration in seconds")
class UserProfile(BaseModel):
"""User profile information."""
+96 -1
View File
@@ -160,6 +160,46 @@ async def list_tools() -> list[Tool]:
"required": ["start_date", "end_date"],
},
),
Tool(
name="query_spo2",
description="Query blood oxygen saturation samples",
inputSchema={
"type": "object",
"properties": {
"start_date": {"type": "string"},
"end_date": {"type": "string"},
"limit": {"type": "integer"},
},
"required": ["start_date", "end_date"],
},
),
Tool(
name="query_stress",
description="Query stress samples",
inputSchema={
"type": "object",
"properties": {
"start_date": {"type": "string"},
"end_date": {"type": "string"},
"level": {"type": "string", "enum": ["low", "medium", "high"]},
"limit": {"type": "integer"},
},
"required": ["start_date", "end_date"],
},
),
Tool(
name="query_abnormal_heart_beat",
description="Query abnormal heart beat events",
inputSchema={
"type": "object",
"properties": {
"start_date": {"type": "string"},
"end_date": {"type": "string"},
"limit": {"type": "integer"},
},
"required": ["start_date", "end_date"],
},
),
Tool(
name="get_data_coverage",
description="Get data coverage",
@@ -192,6 +232,12 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
result = await _handle_query_sleep(arguments)
elif name == "query_workouts":
result = await _handle_query_workouts(arguments)
elif name == "query_spo2":
result = await _handle_query_spo2(arguments)
elif name == "query_stress":
result = await _handle_query_stress(arguments)
elif name == "query_abnormal_heart_beat":
result = await _handle_query_abnormal_heart_beat(arguments)
elif name == "get_data_coverage":
result = await _handle_get_data_coverage(arguments)
else:
@@ -213,7 +259,16 @@ async def _handle_get_connection_status() -> dict:
last_sync = None
available_types = []
if db:
for data_type in ["daily_activity", "heart_rate", "body_measurements", "sleep", "workouts"]:
for data_type in [
"daily_activity",
"heart_rate",
"body_measurements",
"sleep",
"workouts",
"spo2",
"stress",
"abnormal_heart_beat",
]:
state = db.get_sync_state(data_type)
if state and state.get("last_sync_at"):
available_types.append(data_type)
@@ -364,6 +419,46 @@ async def _handle_query_workouts(arguments: dict) -> dict:
).model_dump()
async def _handle_query_spo2(arguments: dict) -> dict:
if not query_service:
return {"status": "error", "error": "Query service not initialized"}
samples = query_service.get_spo2_samples(
start_date=arguments["start_date"],
end_date=arguments["end_date"],
limit=arguments.get("limit"),
)
return QueryResponse(
status="ok", source="cache", data={"samples": samples, "count": len(samples)}
).model_dump()
async def _handle_query_stress(arguments: dict) -> dict:
if not query_service:
return {"status": "error", "error": "Query service not initialized"}
samples = query_service.get_stress_samples(
start_date=arguments["start_date"],
end_date=arguments["end_date"],
level=arguments.get("level"),
limit=arguments.get("limit"),
)
return QueryResponse(
status="ok", source="cache", data={"samples": samples, "count": len(samples)}
).model_dump()
async def _handle_query_abnormal_heart_beat(arguments: dict) -> dict:
if not query_service:
return {"status": "error", "error": "Query service not initialized"}
events = query_service.get_abnormal_heart_beat_events(
start_date=arguments["start_date"],
end_date=arguments["end_date"],
limit=arguments.get("limit"),
)
return QueryResponse(
status="ok", source="cache", data={"events": events, "count": len(events)}
).model_dump()
async def _handle_get_data_coverage(arguments: dict) -> dict:
if not query_service:
return {"status": "error", "error": "Query service not initialized"}
@@ -301,6 +301,62 @@ class QueryService:
return samples[:limit]
return samples
def get_spo2_samples(
self,
start_date: str,
end_date: str,
limit: int | None = None,
) -> list[dict[str, Any]]:
records = self.db.query_spo2_samples(self.user_id, start_date, end_date)
samples = [
{
"timestamp": record["timestamp"],
"spo2_pct": record["spo2_pct"],
}
for record in records
]
return samples[:limit] if limit is not None else samples
def get_stress_samples(
self,
start_date: str,
end_date: str,
level: str | None = None,
limit: int | None = None,
) -> list[dict[str, Any]]:
records = self.db.query_stress_samples(self.user_id, start_date, end_date)
samples = []
for record in records:
if level and record.get("level") != level:
continue
samples.append(
{
"timestamp": record["timestamp"],
"stress_score": record["stress_score"],
"level": record["level"],
}
)
return samples[:limit] if limit is not None else samples
def get_abnormal_heart_beat_events(
self,
start_date: str,
end_date: str,
limit: int | None = None,
) -> list[dict[str, Any]]:
records = self.db.query_abnormal_heart_beat_events(self.user_id, start_date, end_date)
events = [
{
"event_id": record["event_id"],
"start_at": record["start_at"],
"end_at": record["end_at"],
"duration_seconds": record["duration_seconds"],
}
for record in records
]
return events[:limit] if limit is not None else events
def get_data_coverage(self, data_types: list[str] | None = None) -> list[dict[str, Any]]:
"""Get data coverage information."""
coverage = self.db.get_data_coverage(self.user_id)
@@ -123,6 +123,36 @@ class SyncService:
if last_ts is None or sample.timestamp > last_ts:
last_ts = sample.timestamp
elif data_type == "spo2":
records = self.adapter.iter_spo2(start_date, end_date)
async for sample in self._iterate_records(records):
if self.db.insert_spo2_sample(sample):
added += 1
else:
updated += 1
if last_ts is None or sample.timestamp > last_ts:
last_ts = sample.timestamp
elif data_type == "stress":
records = self.adapter.iter_stress(start_date, end_date)
async for sample in self._iterate_records(records):
if self.db.insert_stress_sample(sample):
added += 1
else:
updated += 1
if last_ts is None or sample.timestamp > last_ts:
last_ts = sample.timestamp
elif data_type == "abnormal_heart_beat":
records = self.adapter.iter_abnormal_heart_beat(start_date, end_date)
async for event in self._iterate_records(records):
if self.db.insert_abnormal_heart_beat_event(event):
added += 1
else:
updated += 1
if last_ts is None or event.start_at > last_ts:
last_ts = event.start_at
else:
raise ValueError(f"Unknown data type: {data_type}")
+237
View File
@@ -8,10 +8,13 @@ from pathlib import Path
from typing import Any
from mi_fitness_mcp.models import (
AbnormalHeartBeatEvent,
BodyMeasurement,
DailyActivity,
HeartRateSample,
SleepSession,
SpO2Sample,
StressSample,
Workout,
)
@@ -158,6 +161,63 @@ class Database:
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS spo2_samples (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
source_type TEXT NOT NULL,
source_record_id TEXT,
user_id TEXT NOT NULL,
device_id TEXT,
timezone TEXT DEFAULT 'UTC',
collected_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
timestamp TIMESTAMP NOT NULL,
spo2_pct INTEGER NOT NULL,
UNIQUE(user_id, timestamp)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS stress_samples (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
source_type TEXT NOT NULL,
source_record_id TEXT,
user_id TEXT NOT NULL,
device_id TEXT,
timezone TEXT DEFAULT 'UTC',
collected_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
timestamp TIMESTAMP NOT NULL,
stress_score INTEGER NOT NULL,
level TEXT NOT NULL,
UNIQUE(user_id, timestamp)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS abnormal_heart_beat_events (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
source_type TEXT NOT NULL,
source_record_id TEXT,
user_id TEXT NOT NULL,
device_id TEXT,
timezone TEXT DEFAULT 'UTC',
collected_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
event_id TEXT NOT NULL,
start_at TIMESTAMP NOT NULL,
end_at TIMESTAMP NOT NULL,
duration_seconds INTEGER NOT NULL,
UNIQUE(user_id, event_id)
)
""")
# Sync state table
conn.execute("""
CREATE TABLE IF NOT EXISTS sync_state (
@@ -190,6 +250,18 @@ class Database:
CREATE INDEX IF NOT EXISTS idx_heart_rate_user_ts
ON heart_rate_samples(user_id, timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_spo2_user_ts
ON spo2_samples(user_id, timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_stress_user_ts
ON stress_samples(user_id, timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_abnormal_hr_user_start
ON abnormal_heart_beat_events(user_id, start_at)
""")
conn.commit()
@@ -405,6 +477,97 @@ class Database:
conn.commit()
return cursor.rowcount > 0
def insert_spo2_sample(self, sample: SpO2Sample) -> bool:
with self._get_connection() as conn:
cursor = conn.execute(
"""
INSERT INTO spo2_samples (
id, provider, source_type, source_record_id, user_id, device_id,
timezone, collected_at, timestamp, spo2_pct
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
spo2_pct = excluded.spo2_pct,
updated_at = CURRENT_TIMESTAMP
""",
(
sample.id,
sample.provider,
sample.source_type,
sample.source_record_id,
sample.user_id,
sample.device_id,
sample.timezone,
sample.collected_at.isoformat() if sample.collected_at else None,
sample.timestamp.isoformat(),
sample.spo2_pct,
),
)
conn.commit()
return cursor.rowcount > 0
def insert_stress_sample(self, sample: StressSample) -> bool:
with self._get_connection() as conn:
cursor = conn.execute(
"""
INSERT INTO stress_samples (
id, provider, source_type, source_record_id, user_id, device_id,
timezone, collected_at, timestamp, stress_score, level
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
stress_score = excluded.stress_score,
level = excluded.level,
updated_at = CURRENT_TIMESTAMP
""",
(
sample.id,
sample.provider,
sample.source_type,
sample.source_record_id,
sample.user_id,
sample.device_id,
sample.timezone,
sample.collected_at.isoformat() if sample.collected_at else None,
sample.timestamp.isoformat(),
sample.stress_score,
sample.level,
),
)
conn.commit()
return cursor.rowcount > 0
def insert_abnormal_heart_beat_event(self, event: AbnormalHeartBeatEvent) -> bool:
with self._get_connection() as conn:
cursor = conn.execute(
"""
INSERT INTO abnormal_heart_beat_events (
id, provider, source_type, source_record_id, user_id, device_id,
timezone, collected_at, event_id, start_at, end_at, duration_seconds
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, event_id) DO UPDATE SET
start_at = excluded.start_at,
end_at = excluded.end_at,
duration_seconds = excluded.duration_seconds,
updated_at = CURRENT_TIMESTAMP
""",
(
event.id,
event.provider,
event.source_type,
event.source_record_id,
event.user_id,
event.device_id,
event.timezone,
event.collected_at.isoformat() if event.collected_at else None,
event.event_id,
event.start_at.isoformat(),
event.end_at.isoformat(),
event.duration_seconds,
),
)
conn.commit()
return cursor.rowcount > 0
def update_sync_state(self, data_type: str, last_record_ts: datetime | None = None) -> None:
"""Update sync state for a data type."""
with self._get_connection() as conn:
@@ -522,6 +685,60 @@ class Database:
).fetchall()
return [dict(row) for row in rows]
def query_spo2_samples(
self,
user_id: str,
start_date: str,
end_date: str,
) -> list[dict[str, Any]]:
with self._get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM spo2_samples
WHERE user_id = ?
AND date(timestamp) >= ? AND date(timestamp) <= ?
ORDER BY timestamp
""",
(user_id, start_date, end_date),
).fetchall()
return [dict(row) for row in rows]
def query_stress_samples(
self,
user_id: str,
start_date: str,
end_date: str,
) -> list[dict[str, Any]]:
with self._get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM stress_samples
WHERE user_id = ?
AND date(timestamp) >= ? AND date(timestamp) <= ?
ORDER BY timestamp
""",
(user_id, start_date, end_date),
).fetchall()
return [dict(row) for row in rows]
def query_abnormal_heart_beat_events(
self,
user_id: str,
start_date: str,
end_date: str,
) -> list[dict[str, Any]]:
with self._get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM abnormal_heart_beat_events
WHERE user_id = ?
AND date(start_at) >= ? AND date(start_at) <= ?
ORDER BY start_at
""",
(user_id, start_date, end_date),
).fetchall()
return [dict(row) for row in rows]
def get_data_coverage(self, user_id: str) -> list[dict[str, Any]]:
"""Get data coverage statistics."""
with self._get_connection() as conn:
@@ -626,4 +843,24 @@ class Database:
}
)
for data_type, table_name, date_expr in [
("spo2", "spo2_samples", "timestamp"),
("stress", "stress_samples", "timestamp"),
("abnormal_heart_beat", "abnormal_heart_beat_events", "start_at"),
]:
row = conn.execute(
f"""
SELECT
MIN(date({date_expr})) as first_date,
MAX(date({date_expr})) as last_date,
COUNT(DISTINCT date({date_expr})) as days_with_data
FROM {table_name}
WHERE user_id = ?
""",
(user_id,),
).fetchone()
if row and row["first_date"]:
results.append({"data_type": data_type, **dict(row)})
return results