From 6771fc841c5c4a36d5b78c4cf222d6d3c56cc53e Mon Sep 17 00:00:00 2001 From: Aleksej Kubulashvili Date: Thu, 12 Mar 2026 10:03:58 +0300 Subject: [PATCH] Initialize standalone Mi Fitness MCP repository --- .gitignore | 29 + README.md | 40 ++ mcp-server-config.example.json | 8 + probe_mifitness.py | 268 ++++++++ pyproject.toml | 57 ++ src/mi_fitness_mcp/__init__.py | 3 + src/mi_fitness_mcp/adapters/base.py | 87 +++ .../adapters/mi_fitness_cloud.py | 349 ++++++++++ src/mi_fitness_mcp/auth/__init__.py | 32 + src/mi_fitness_mcp/cli_mi_fitness.py | 5 + src/mi_fitness_mcp/config.py | 71 ++ src/mi_fitness_mcp/main.py | 179 +++++ src/mi_fitness_mcp/models/__init__.py | 183 ++++++ src/mi_fitness_mcp/server.py | 181 +++++ src/mi_fitness_mcp/services/__init__.py | 6 + src/mi_fitness_mcp/services/autosync.py | 223 +++++++ src/mi_fitness_mcp/services/query_service.py | 301 +++++++++ src/mi_fitness_mcp/services/sync_service.py | 164 +++++ src/mi_fitness_mcp/storage/__init__.py | 619 ++++++++++++++++++ 19 files changed, 2805 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 mcp-server-config.example.json create mode 100644 probe_mifitness.py create mode 100644 pyproject.toml create mode 100644 src/mi_fitness_mcp/__init__.py create mode 100644 src/mi_fitness_mcp/adapters/base.py create mode 100644 src/mi_fitness_mcp/adapters/mi_fitness_cloud.py create mode 100644 src/mi_fitness_mcp/auth/__init__.py create mode 100644 src/mi_fitness_mcp/cli_mi_fitness.py create mode 100644 src/mi_fitness_mcp/config.py create mode 100644 src/mi_fitness_mcp/main.py create mode 100644 src/mi_fitness_mcp/models/__init__.py create mode 100644 src/mi_fitness_mcp/server.py create mode 100644 src/mi_fitness_mcp/services/__init__.py create mode 100644 src/mi_fitness_mcp/services/autosync.py create mode 100644 src/mi_fitness_mcp/services/query_service.py create mode 100644 src/mi_fitness_mcp/services/sync_service.py create mode 100644 src/mi_fitness_mcp/storage/__init__.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8dd5c5f --- /dev/null +++ b/.gitignore @@ -0,0 +1,29 @@ +.venv/ +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.so +*.egg +*.egg-info/ +dist/ +build/ +*.db +*.sqlite3 +.DS_Store +.env +.env.local +*.log +.pytest_cache/ +.coverage +htmlcov/ +.tox/ +.mypy_cache/ +.dmypy.json +*.iml +.idea/ +.vscode/ +*.swp +*.swo +*~ diff --git a/README.md b/README.md new file mode 100644 index 0000000..60a6439 --- /dev/null +++ b/README.md @@ -0,0 +1,40 @@ +# Mi Fitness MCP + +Отдельный MCP-сервер для `Mi Fitness`. + +Подтверждённые данные: + +- шаги +- калории +- пульс +- вес и состав тела + +## Установка + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -e "." +``` + +## Команды + +```bash +mi-fitness-mcp setup --mode mi_fitness_cloud --user-id "" --pass-token "" --region ru +mi-fitness-mcp doctor +mi-fitness-mcp sync --start-date 2025-04-01 --end-date 2025-05-31 +mi-fitness-mcp serve +``` + +## Настройка Claude Desktop + +```json +{ + "mcpServers": { + "mi-fitness": { + "command": "mi-fitness-mcp", + "args": ["serve"] + } + } +} +``` diff --git a/mcp-server-config.example.json b/mcp-server-config.example.json new file mode 100644 index 0000000..b06f7eb --- /dev/null +++ b/mcp-server-config.example.json @@ -0,0 +1,8 @@ +{ + "mcpServers": { + "mi-fitness": { + "command": "mi-fitness-mcp", + "args": ["serve"] + } + } +} diff --git a/probe_mifitness.py b/probe_mifitness.py new file mode 100644 index 0000000..be48785 --- /dev/null +++ b/probe_mifitness.py @@ -0,0 +1,268 @@ +import argparse +import base64 +import getpass +import hashlib +import json +import os +import random +import struct +from datetime import datetime +from urllib.parse import urlencode + +import httpx + + +LOGIN_PREFIX = b"&&&START&&&" +DEFAULT_KEYS = [ + "weight", + "steps", + "step", + "activity", + "sleep", + "heart_rate", + "heartrate", + "hr", + "sport", + "workout", + "calories", +] +DEFAULT_REGIONS = ["cn", "de", "i2", "ru", "sg", "us"] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(prog="probe_mifitness") + parser.add_argument("--username") + parser.add_argument("--password") + parser.add_argument("--token") + parser.add_argument("--user-id") + parser.add_argument("--pass-token") + parser.add_argument("--start-date", default="2025-04-01") + parser.add_argument("--end-date", default="2025-05-31") + parser.add_argument("--regions", nargs="*", default=DEFAULT_REGIONS) + parser.add_argument("--keys", nargs="*", default=DEFAULT_KEYS) + return parser.parse_args() + + +def md5_upper(text: str) -> str: + return hashlib.md5(text.encode()).hexdigest().upper() + + +def read_login_payload(text: str) -> dict: + body = text.encode() + if not body.startswith(LOGIN_PREFIX): + raise RuntimeError("unexpected Xiaomi login response") + return json.loads(body[len(LOGIN_PREFIX):].decode()) + + +def rc4_crypt(key: bytes, payload: bytes) -> bytes: + s = list(range(256)) + j = 0 + key_len = len(key) + for i in range(256): + j = (j + s[i] + key[i % key_len]) % 256 + s[i], s[j] = s[j], s[i] + i = 0 + j = 0 + + def next_byte() -> int: + nonlocal i, j + i = (i + 1) % 256 + j = (j + s[i]) % 256 + s[i], s[j] = s[j], s[i] + return s[(s[i] + s[j]) % 256] + + for _ in range(1024): + next_byte() + + out = bytearray() + for b in payload: + out.append(b ^ next_byte()) + return bytes(out) + + +def gen_nonce() -> bytes: + raw = bytearray(os.urandom(8)) + raw.extend(struct.pack(">I", int(datetime.now().timestamp() // 60))) + return bytes(raw) + + +def gen_signed_nonce(ssecurity: bytes, nonce: bytes) -> bytes: + return hashlib.sha256(ssecurity + nonce).digest() + + +def gen_signature(method: str, path: str, values: dict[str, str], signed_nonce: bytes) -> str: + base = method + "&" + path + "&data=" + values["data"] + if "rc4_hash__" in values: + base += "&rc4_hash__=" + values["rc4_hash__"] + base += "&" + base64.b64encode(signed_nonce).decode() + return base64.b64encode(hashlib.sha1(base.encode()).digest()).decode() + + +class MiFitnessProbe: + def __init__(self) -> None: + self.client = httpx.Client(timeout=30.0, follow_redirects=False) + self.sid = "miothealth" + self.cookies = "" + self.ssecurity = b"" + self.pass_token = "" + self.user_id = 0 + + def login(self, username: str, password: str) -> None: + info = self.service_login() + auth = self.service_login2(info, username, password) + self.service_login3(auth["location"]) + + def login_with_token(self, token: str | None = None, user_id: str | None = None, pass_token: str | None = None) -> None: + if token: + user_id, pass_token = token.split(":", 1) + if not user_id or not pass_token: + raise RuntimeError("token login requires user_id and pass_token") + + response = self.client.get( + f"https://account.xiaomi.com/pass/serviceLogin?_json=true&sid={self.sid}", + headers={"Cookie": f"userId={user_id}; passToken={pass_token}"}, + ) + response.raise_for_status() + payload = read_login_payload(response.text) + if "passToken" not in payload or "ssecurity" not in payload or "userId" not in payload or "location" not in payload: + details = { + "keys": sorted(payload.keys()), + "code": payload.get("code"), + "description": payload.get("description"), + "desc": payload.get("desc"), + "location": payload.get("location"), + } + raise RuntimeError(f"xiaomi token login failed; details={json.dumps(details, ensure_ascii=False)}") + self.pass_token = payload["passToken"] + self.ssecurity = base64.b64decode(payload["ssecurity"]) + self.user_id = int(payload["userId"]) + self.service_login3(payload["location"]) + + def service_login(self) -> dict: + response = self.client.get(f"https://account.xiaomi.com/pass/serviceLogin?_json=true&sid={self.sid}") + response.raise_for_status() + return read_login_payload(response.text) + + def service_login2(self, info: dict, username: str, password: str) -> dict: + form = { + "_json": "true", + "hash": md5_upper(password), + "sid": info["sid"], + "callback": info["callback"], + "_sign": info["_sign"], + "qs": info["qs"], + "user": username, + } + device_id = "".join(random.choice("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") for _ in range(16)) + response = self.client.post( + "https://account.xiaomi.com/pass/serviceLoginAuth2", + headers={ + "Cookie": f"deviceId={device_id}", + "Content-Type": "application/x-www-form-urlencoded", + }, + content=urlencode(form), + ) + response.raise_for_status() + payload = read_login_payload(response.text) + if "passToken" not in payload or "ssecurity" not in payload or "userId" not in payload: + message = payload.get("description") or payload.get("desc") or payload.get("message") + details = { + "keys": sorted(payload.keys()), + "code": payload.get("code"), + "description": payload.get("description"), + "desc": payload.get("desc"), + "captchaUrl": payload.get("captchaUrl"), + "notificationUrl": payload.get("notificationUrl"), + "location": payload.get("location"), + } + raise RuntimeError(f"xiaomi login failed: {message or 'missing auth tokens'}; details={json.dumps(details, ensure_ascii=False)}") + self.pass_token = payload["passToken"] + self.ssecurity = base64.b64decode(payload["ssecurity"]) + self.user_id = int(payload["userId"]) + return payload + + def service_login3(self, location: str) -> None: + response = self.client.get(location) + response.raise_for_status() + cookie_parts = [] + for value in response.headers.get_list("set-cookie"): + cookie_parts.append(value.split(";", 1)[0]) + self.cookies = "; ".join(cookie_parts) + + def request(self, base_url: str, api_path: str, params: dict) -> dict: + form = {"data": json.dumps(params, separators=(",", ":"))} + nonce = gen_nonce() + signed_nonce = gen_signed_nonce(self.ssecurity, nonce) + form["rc4_hash__"] = gen_signature("POST", api_path, form, signed_nonce) + + encrypted = {} + for key, value in form.items(): + encrypted[key] = base64.b64encode(rc4_crypt(signed_nonce, value.encode())).decode() + + encrypted["signature"] = gen_signature("POST", api_path, encrypted, signed_nonce) + encrypted["_nonce"] = base64.b64encode(nonce).decode() + + response = self.client.post( + base_url + api_path, + headers={ + "Cookie": self.cookies, + "Content-Type": "application/x-www-form-urlencoded", + }, + content=urlencode(encrypted), + ) + response.raise_for_status() + plaintext = rc4_crypt(signed_nonce, base64.b64decode(response.text)) + payload = json.loads(plaintext) + if payload.get("code") != 0: + raise RuntimeError(payload.get("message", "unknown mi fitness error")) + return payload.get("result", {}) + + +def region_base(region: str) -> str: + if region in ("", "cn"): + return "https://hlth.io.mi.com" + return f"https://{region}.hlth.io.mi.com" + + +def sample_view(items: list[dict]) -> str: + if not items: + return "[]" + return json.dumps(items[:2], ensure_ascii=False)[:700] + + +def run_probe(args: argparse.Namespace) -> None: + probe = MiFitnessProbe() + + if args.token or (args.user_id and args.pass_token): + probe.login_with_token(token=args.token, user_id=args.user_id, pass_token=args.pass_token) + else: + username = args.username or input("Xiaomi login (email/phone): ").strip() + password = args.password or getpass.getpass("Xiaomi password: ") + probe.login(username, password) + + print(f"login ok: user_id={probe.user_id}") + + for region in args.regions: + base = region_base(region) + print(f"\n=== region={region} base={base} ===") + for key in args.keys: + try: + result = probe.request( + base, + "/app/v1/data/get_fitness_data_by_time", + { + "start_time": int(datetime.fromisoformat(args.start_date).timestamp()), + "end_time": int(datetime.fromisoformat(args.end_date + "T23:59:59").timestamp()), + "key": key, + }, + ) + data_list = result.get("data_list", []) + print(f"key={key} count={len(data_list)} has_more={result.get('has_more')} next_key={bool(result.get('next_key'))}") + if data_list: + print(sample_view(data_list)) + except Exception as exc: + print(f"key={key} error={exc}") + + +if __name__ == "__main__": + run_probe(parse_args()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d3973b9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,57 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "mi-fitness-mcp" +version = "0.1.0" +description = "MCP server for Mi Fitness data access" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.11" +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: End Users/Desktop", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] +dependencies = [ + "mcp>=1.0.0", + "pydantic>=2.0.0", + "httpx>=0.27.0", + "platformdirs>=4.0.0", + "keyring>=25.0.0", + "click>=8.0.0", + "rich>=13.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", + "respx>=0.21.0", + "ruff>=0.4.0", +] + +[project.scripts] +mi-fitness-mcp = "mi_fitness_mcp.cli_mi_fitness:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/mi_fitness_mcp"] + +[tool.ruff] +target-version = "py311" +line-length = 100 +select = ["E", "F", "I", "N", "W", "UP", "B", "C4", "SIM"] +ignore = ["E501"] + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" + +[tool.pytest.ini_options] +testpaths = ["tests"] +asyncio_mode = "auto" diff --git a/src/mi_fitness_mcp/__init__.py b/src/mi_fitness_mcp/__init__.py new file mode 100644 index 0000000..286ab7c --- /dev/null +++ b/src/mi_fitness_mcp/__init__.py @@ -0,0 +1,3 @@ +"""Mi Fitness MCP server.""" + +__version__ = "0.1.0" diff --git a/src/mi_fitness_mcp/adapters/base.py b/src/mi_fitness_mcp/adapters/base.py new file mode 100644 index 0000000..ef4cbed --- /dev/null +++ b/src/mi_fitness_mcp/adapters/base.py @@ -0,0 +1,87 @@ +"""Base adapter interface for data sources.""" + +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator, Iterator +from pathlib import Path +from typing import Any, Coroutine, Union + +from mi_fitness_mcp.models import ( + BodyMeasurement, + DailyActivity, + HeartRateSample, + SleepSession, + Workout, +) + + +class DataAdapter(ABC): + """Abstract base class for data source adapters.""" + + @abstractmethod + def connect(self) -> Union[bool, "Coroutine[Any, Any, bool]"]: + """Connect to data source. + + Returns: + True if connection successful + """ + pass + + @abstractmethod + def is_connected(self) -> bool: + """Check if connected to data source.""" + pass + + @abstractmethod + def get_user_id(self) -> str | None: + """Get user identifier from data source.""" + pass + + @abstractmethod + def iter_daily_activity( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> Union[Iterator[DailyActivity], AsyncIterator[DailyActivity]]: + """Iterate over daily activity records.""" + pass + + @abstractmethod + def iter_sleep_sessions( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> Union[Iterator[SleepSession], AsyncIterator[SleepSession]]: + """Iterate over sleep session records.""" + pass + + @abstractmethod + def iter_workouts( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> Union[Iterator[Workout], AsyncIterator[Workout]]: + """Iterate over workout records.""" + pass + + @abstractmethod + def iter_body_measurements( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> Union[Iterator[BodyMeasurement], AsyncIterator[BodyMeasurement]]: + """Iterate over body measurement records.""" + pass + + @abstractmethod + def iter_heart_rate( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> Union[Iterator[HeartRateSample], AsyncIterator[HeartRateSample]]: + """Iterate over heart rate records.""" + pass + + @abstractmethod + def get_available_data_types(self) -> list[str]: + """Get list of available data types.""" + pass diff --git a/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py b/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py new file mode 100644 index 0000000..8890c26 --- /dev/null +++ b/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py @@ -0,0 +1,349 @@ +import base64 +import hashlib +import json +import os +import random +import struct +from collections import defaultdict +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from typing import Any +from urllib.parse import urlencode + +import httpx + +from mi_fitness_mcp.adapters.base import DataAdapter +from mi_fitness_mcp.models import BodyMeasurement, DailyActivity, HeartRateSample, SleepSession, Workout + + +LOGIN_PREFIX = b"&&&START&&&" +KNOWN_REGIONS = ["ru", "cn", "de", "i2", "sg", "us"] + + +def _read_login_payload(text: str) -> dict: + payload = text.encode() + if not payload.startswith(LOGIN_PREFIX): + raise RuntimeError("unexpected Xiaomi login response") + return json.loads(payload[len(LOGIN_PREFIX):].decode()) + + +def _rc4_crypt(key: bytes, payload: bytes) -> bytes: + s = list(range(256)) + j = 0 + key_len = len(key) + for i in range(256): + j = (j + s[i] + key[i % key_len]) % 256 + s[i], s[j] = s[j], s[i] + i = 0 + j = 0 + + def next_byte() -> int: + nonlocal i, j + i = (i + 1) % 256 + j = (j + s[i]) % 256 + s[i], s[j] = s[j], s[i] + return s[(s[i] + s[j]) % 256] + + for _ in range(1024): + next_byte() + + output = bytearray() + for value in payload: + output.append(value ^ next_byte()) + return bytes(output) + + +def _gen_nonce() -> bytes: + raw = bytearray(os.urandom(8)) + raw.extend(struct.pack(">I", int(datetime.now().timestamp() // 60))) + return bytes(raw) + + +def _gen_signed_nonce(ssecurity: bytes, nonce: bytes) -> bytes: + return hashlib.sha256(ssecurity + nonce).digest() + + +def _gen_signature(method: str, path: str, values: dict[str, str], signed_nonce: bytes) -> str: + base = method + "&" + path + "&data=" + values["data"] + if "rc4_hash__" in values: + base += "&rc4_hash__=" + values["rc4_hash__"] + base += "&" + base64.b64encode(signed_nonce).decode() + return base64.b64encode(hashlib.sha1(base.encode()).digest()).decode() + + +class MiFitnessCloudAdapter(DataAdapter): + def __init__(self, user_id: str | None = None, pass_token: str | None = None, region: str = "ru"): + self.user_id = user_id + self.pass_token = pass_token + self.region = region + self._cookies = "" + self._ssecurity = b"" + self._client: httpx.AsyncClient | None = None + self._connected = False + self._available_types: list[str] = [] + + async def connect(self) -> bool: + if not self.user_id or not self.pass_token: + return False + + self._client = httpx.AsyncClient(timeout=30.0, follow_redirects=False) + + try: + await self._login_with_token(self.user_id, self.pass_token) + self.region = await self._discover_region(self.region) + self._available_types = await self._discover_data_types() + self._connected = True + return True + except Exception: + self._connected = False + return False + + def is_connected(self) -> bool: + return self._connected and self._client is not None + + def get_user_id(self) -> str | None: + return self.user_id + + def get_available_data_types(self) -> list[str]: + return self._available_types.copy() + + async def _login_with_token(self, user_id: str, pass_token: str) -> None: + if not self._client: + raise RuntimeError("client not initialized") + + response = await self._client.get( + "https://account.xiaomi.com/pass/serviceLogin?_json=true&sid=miothealth", + headers={"Cookie": f"userId={user_id}; passToken={pass_token}"}, + ) + response.raise_for_status() + payload = _read_login_payload(response.text) + self.pass_token = payload["passToken"] + self.user_id = str(payload["userId"]) + self._ssecurity = base64.b64decode(payload["ssecurity"]) + + redirect = await self._client.get(payload["location"]) + redirect.raise_for_status() + cookie_parts = [value.split(";", 1)[0] for value in redirect.headers.get_list("set-cookie")] + self._cookies = "; ".join(cookie_parts) + + async def _request(self, base_url: str, api_path: str, payload: dict) -> dict: + if not self._client: + raise RuntimeError("client not initialized") + + form = {"data": json.dumps(payload, separators=(",", ":"))} + nonce = _gen_nonce() + signed_nonce = _gen_signed_nonce(self._ssecurity, nonce) + form["rc4_hash__"] = _gen_signature("POST", api_path, form, signed_nonce) + + encrypted: dict[str, str] = {} + for key, value in form.items(): + encrypted[key] = base64.b64encode(_rc4_crypt(signed_nonce, value.encode())).decode() + + encrypted["signature"] = _gen_signature("POST", api_path, encrypted, signed_nonce) + encrypted["_nonce"] = base64.b64encode(nonce).decode() + + response = await self._client.post( + base_url + api_path, + headers={ + "Cookie": self._cookies, + "Content-Type": "application/x-www-form-urlencoded", + }, + content=urlencode(encrypted), + ) + response.raise_for_status() + plaintext = _rc4_crypt(signed_nonce, base64.b64decode(response.text)) + body = json.loads(plaintext) + if body.get("code") != 0: + raise RuntimeError(body.get("message", "unknown mi fitness error")) + return body.get("result", {}) + + async def _fetch_key(self, key: str, start_date: str, end_date: str, region: str | None = None) -> list[dict]: + region_name = region or self.region + base_url = "https://hlth.io.mi.com" if region_name in ("", "cn") else f"https://{region_name}.hlth.io.mi.com" + start_time = int(datetime.fromisoformat(start_date).replace(tzinfo=UTC).timestamp()) + end_time = int(datetime.fromisoformat(end_date + "T23:59:59").replace(tzinfo=UTC).timestamp()) + next_key = None + items: list[dict] = [] + + while True: + payload = { + "start_time": start_time, + "end_time": end_time, + "key": key, + } + if next_key: + payload["next_key"] = next_key + + result = await self._request(base_url, "/app/v1/data/get_fitness_data_by_time", payload) + items.extend(result.get("data_list", [])) + if not result.get("has_more") or not result.get("next_key"): + break + next_key = result.get("next_key") + + return items + + async def _discover_region(self, preferred_region: str) -> str: + candidates = [preferred_region] + [region for region in KNOWN_REGIONS if region != preferred_region] + for region in candidates: + for key in ("weight", "steps", "heart_rate"): + try: + result = await self._fetch_key(key, "2025-04-01", "2025-05-31", region=region) + if result: + return region + except Exception: + continue + return preferred_region + + async def _discover_data_types(self) -> list[str]: + types: list[str] = [] + probes = { + "daily_activity": "steps", + "heart_rate": "heart_rate", + "body_measurements": "weight", + } + for data_type, key in probes.items(): + try: + items = await self._fetch_key(key, "2025-04-01", "2025-05-31") + if items: + types.append(data_type) + except Exception: + continue + return types + + def _record_datetime(self, item: dict) -> datetime: + timestamp = int(item.get("time", 0)) + zone_offset = int(item.get("zone_offset", 0)) + return datetime.fromtimestamp(timestamp + zone_offset, tz=UTC).replace(tzinfo=None) + + def _parse_value(self, item: dict) -> dict[str, Any]: + raw = item.get("value", "{}") + if isinstance(raw, dict): + return raw + return json.loads(raw) + + def _optional_float(self, value: Any) -> float | None: + if value is None: + return None + parsed = float(value) + return None if parsed == 0 else parsed + + def _optional_int(self, value: Any) -> int | None: + if value is None: + return None + parsed = int(float(value)) + return None if parsed == 0 else parsed + + async def iter_daily_activity( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> AsyncIterator[DailyActivity]: + if not self.is_connected() or not start_date or not end_date: + return + yield + + records = await self._fetch_key("steps", start_date, end_date) + daily: dict[str, dict[str, float]] = defaultdict(lambda: {"steps": 0, "distance_m": 0.0, "active_kcal": 0.0}) + for item in records: + payload = self._parse_value(item) + date_str = self._record_datetime(item).strftime("%Y-%m-%d") + daily[date_str]["steps"] += int(payload.get("steps", 0)) + daily[date_str]["distance_m"] += float(payload.get("distance", 0)) + daily[date_str]["active_kcal"] += float(payload.get("calories", 0)) + + calorie_records = await self._fetch_key("calories", start_date, end_date) + calorie_totals: dict[str, float] = defaultdict(float) + for item in calorie_records: + payload = self._parse_value(item) + date_str = self._record_datetime(item).strftime("%Y-%m-%d") + calorie_totals[date_str] += float(payload.get("calories", 0)) + + for date_str, total in calorie_totals.items(): + daily[date_str]["active_kcal"] = total + + for date_str, values in sorted(daily.items()): + yield DailyActivity( + id=f"mi_fitness_activity_{date_str}", + provider="mi_fitness", + source_type="cloud_session", + user_id=self.user_id or "unknown", + date=date_str, + steps=int(values["steps"]), + distance_m=values["distance_m"], + active_kcal=values["active_kcal"], + ) + + async def iter_sleep_sessions( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> AsyncIterator[SleepSession]: + if False: + yield + + async def iter_workouts( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> AsyncIterator[Workout]: + if False: + yield + + async def iter_body_measurements( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> AsyncIterator[BodyMeasurement]: + if not self.is_connected() or not start_date or not end_date: + return + yield + + records = await self._fetch_key("weight", start_date, end_date) + for item in records: + payload = self._parse_value(item) + measured_at = self._record_datetime(item) + yield BodyMeasurement( + id=f"mi_fitness_weight_{int(item.get('time', 0))}", + provider="mi_fitness", + source_type="cloud_session", + user_id=self.user_id or "unknown", + timestamp=measured_at, + weight_kg=float(payload.get("weight", 0)), + bmi=float(payload["bmi"]) if payload.get("bmi") is not None else None, + body_fat_pct=self._optional_float(payload.get("body_fat_rate")), + muscle_mass_kg=self._optional_float(payload.get("muscle_rate")), + water_pct=self._optional_float(payload.get("moisture_rate")), + bone_mass_kg=self._optional_float(payload.get("bone_mass")), + visceral_fat_score=self._optional_int(payload.get("visceral_fat")), + basal_metabolism_kcal=self._optional_int(payload.get("basal_metabolism")), + metabolic_age=self._optional_int(payload.get("body_age")), + ) + + async def iter_heart_rate( + self, + start_date: str | None = None, + end_date: str | None = None, + ) -> AsyncIterator[HeartRateSample]: + if not self.is_connected() or not start_date or not end_date: + return + yield + + records = await self._fetch_key("heart_rate", start_date, end_date) + for item in records: + payload = self._parse_value(item) + sample_type = "passive" if int(payload.get("type", 0)) == 0 else "active" + yield HeartRateSample( + id=f"mi_fitness_hr_{int(item.get('time', 0))}", + provider="mi_fitness", + source_type="cloud_session", + user_id=self.user_id or "unknown", + timestamp=self._record_datetime(item), + bpm=int(payload.get("bpm", 0)), + sample_type=sample_type, + ) + + async def close(self) -> None: + if self._client: + await self._client.aclose() + self._client = None + self._connected = False diff --git a/src/mi_fitness_mcp/auth/__init__.py b/src/mi_fitness_mcp/auth/__init__.py new file mode 100644 index 0000000..399665e --- /dev/null +++ b/src/mi_fitness_mcp/auth/__init__.py @@ -0,0 +1,32 @@ +"""Authentication helpers for Mi Fitness cloud API.""" + +import keyring + + +SERVICE_NAME = "mi-fitness-mcp" +ACCOUNT_NAME = "mi_fitness_auth" + + +def save_mi_fitness_token(user_id: str, pass_token: str) -> None: + keyring.set_password(SERVICE_NAME, f"{ACCOUNT_NAME}_user_id", user_id) + keyring.set_password(SERVICE_NAME, f"{ACCOUNT_NAME}_pass_token", pass_token) + + +def load_mi_fitness_token() -> tuple[str | None, str | None]: + try: + user_id = keyring.get_password(SERVICE_NAME, f"{ACCOUNT_NAME}_user_id") + pass_token = keyring.get_password(SERVICE_NAME, f"{ACCOUNT_NAME}_pass_token") + return user_id, pass_token + except Exception: + return None, None + + +def delete_mi_fitness_token() -> None: + try: + keyring.delete_password(SERVICE_NAME, f"{ACCOUNT_NAME}_user_id") + except keyring.errors.PasswordDeleteError: + pass + try: + keyring.delete_password(SERVICE_NAME, f"{ACCOUNT_NAME}_pass_token") + except keyring.errors.PasswordDeleteError: + pass diff --git a/src/mi_fitness_mcp/cli_mi_fitness.py b/src/mi_fitness_mcp/cli_mi_fitness.py new file mode 100644 index 0000000..9741ab3 --- /dev/null +++ b/src/mi_fitness_mcp/cli_mi_fitness.py @@ -0,0 +1,5 @@ +from mi_fitness_mcp.main import main as _main + + +def main() -> None: + _main() diff --git a/src/mi_fitness_mcp/config.py b/src/mi_fitness_mcp/config.py new file mode 100644 index 0000000..190fa90 --- /dev/null +++ b/src/mi_fitness_mcp/config.py @@ -0,0 +1,71 @@ +"""Configuration management for Mi Fitness MCP.""" + +import json +from pathlib import Path +from typing import Literal + +from platformdirs import user_config_dir, user_data_dir, user_log_dir +from pydantic import BaseModel, Field + + +def _default_database_path() -> Path: + return Path(user_data_dir("mi-fitness-mcp")) / "mi_fitness.db" + + +def _default_logs_path() -> Path: + return Path(user_log_dir("mi-fitness-mcp")) / "mi_fitness.log" + + +class Config(BaseModel): + mode: Literal["mi_fitness_cloud", "not_configured"] = "not_configured" + region: str = "ru" + timezone: str = Field(default="UTC") + database_path: Path = Field(default_factory=_default_database_path) + logs_path: Path = Field(default_factory=_default_logs_path) + auto_sync_on_start: bool = True + stale_after_minutes: int = 60 + store_raw_payloads: bool = True + default_lookback_days: int = 30 + + class Config: + arbitrary_types_allowed = True + + +def get_config_dir() -> Path: + config_dir = Path(user_config_dir("mi-fitness-mcp")) + config_dir.mkdir(parents=True, exist_ok=True) + return config_dir + + +def get_config_path() -> Path: + return get_config_dir() / "config.json" + + +def load_config() -> Config: + config_path = get_config_path() + if not config_path.exists(): + config = Config() + save_config(config) + return config + + with open(config_path, "r", encoding="utf-8") as f: + data = json.load(f) + + if "database_path" in data and isinstance(data["database_path"], str): + data["database_path"] = Path(data["database_path"]) + if "logs_path" in data and isinstance(data["logs_path"], str): + data["logs_path"] = Path(data["logs_path"]) + + return Config(**data) + + +def save_config(config: Config) -> None: + config_path = get_config_path() + config_path.parent.mkdir(parents=True, exist_ok=True) + + data = config.model_dump() + data["database_path"] = str(data["database_path"]) + data["logs_path"] = str(data["logs_path"]) + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) diff --git a/src/mi_fitness_mcp/main.py b/src/mi_fitness_mcp/main.py new file mode 100644 index 0000000..94b9ca6 --- /dev/null +++ b/src/mi_fitness_mcp/main.py @@ -0,0 +1,179 @@ +"""CLI for Mi Fitness MCP.""" + +import argparse +import asyncio +import sys + +from mi_fitness_mcp.adapters.mi_fitness_cloud import MiFitnessCloudAdapter +from mi_fitness_mcp.auth import load_mi_fitness_token, save_mi_fitness_token +from mi_fitness_mcp.config import Config, get_config_path, load_config, save_config +from mi_fitness_mcp.server import main as server_main +from mi_fitness_mcp.storage import Database +from mi_fitness_mcp.services.sync_service import SyncService + + +PROGRAM_NAME = "mi-fitness-mcp" + + +async def _check_adapter_health(adapter) -> tuple[bool, list[str], str | None]: + connected = await adapter.connect() + region = getattr(adapter, "region", None) + data_types = adapter.get_available_data_types() if connected else [] + if hasattr(adapter, "close"): + await adapter.close() + return connected, data_types, region + + +def cmd_setup(args): + if args.mode == "mi_fitness_cloud" and args.user_id and args.pass_token: + save_mi_fitness_token(args.user_id, args.pass_token) + config = Config(mode="mi_fitness_cloud", region=args.region or "ru") + save_config(config) + print("✅ Конфигурация сохранена (mi_fitness_cloud)") + print(f" User ID: {args.user_id}") + print(f" Region: {config.region}") + return + + print(f"{PROGRAM_NAME} - Setup Wizard") + print("=" * 50) + print() + user_id = input("Mi Fitness user_id: ").strip() + pass_token = input("Mi Fitness passToken: ").strip() + region = input("Region [ru]: ").strip() or "ru" + if not user_id or not pass_token: + print("❌ Требуются user_id и passToken") + sys.exit(1) + save_mi_fitness_token(user_id, pass_token) + config = Config(mode="mi_fitness_cloud", region=region) + save_config(config) + print("✅ Mi Fitness конфигурация сохранена!") + + +def cmd_doctor(args): + print(f"{PROGRAM_NAME} - Doctor") + print("=" * 50) + print() + config_path = get_config_path() + print(f"Конфигурация: {config_path}") + + if not config_path.exists(): + print("❌ Конфигурация не найдена") + print(f" Запустите: {PROGRAM_NAME} setup") + sys.exit(1) + + try: + config = load_config() + print("✅ Конфигурация загружена") + print(f" Режим: {config.mode}") + user_id, pass_token = load_mi_fitness_token() + if user_id and pass_token: + print("✅ Mi Fitness credentials найдены") + print(f" User ID: {user_id}") + print(f" Region: {config.region}") + adapter = MiFitnessCloudAdapter(user_id=user_id, pass_token=pass_token, region=config.region) + connected, data_types, region = asyncio.run(_check_adapter_health(adapter)) + print(f" Подключение: {'✅' if connected else '❌'}") + if connected: + print(f" Определённый регион: {region}") + print(f" Типы данных: {', '.join(data_types)}") + else: + print("❌ Mi Fitness credentials не найдены") + print(f" Запустите: {PROGRAM_NAME} setup") + + print() + print(f"База данных: {config.database_path}") + if config.database_path.exists(): + Database(config.database_path) + print("✅ База данных доступна") + else: + print("ℹ️ База данных будет создана при первом запуске") + except Exception as e: + print(f"❌ Ошибка загрузки конфигурации: {e}") + sys.exit(1) + + +async def cmd_sync_async(args): + print(f"{PROGRAM_NAME} - Sync") + print("=" * 50) + print() + + try: + config = load_config() + except Exception as e: + print(f"❌ Ошибка загрузки конфигурации: {e}") + sys.exit(1) + + if config.mode == "not_configured": + print("❌ Сервер не настроен") + print(f" Запустите: {PROGRAM_NAME} setup") + sys.exit(1) + + user_id, pass_token = load_mi_fitness_token() + if not user_id or not pass_token: + print("❌ Mi Fitness credentials не найдены") + print(f" Запустите: {PROGRAM_NAME} setup") + sys.exit(1) + + db = Database(config.database_path) + adapter = MiFitnessCloudAdapter(user_id=user_id, pass_token=pass_token, region=config.region) + if not await adapter.connect(): + print("❌ Не удалось подключиться к Mi Fitness API") + sys.exit(1) + + sync_service = SyncService(adapter, db) + data_types = [args.type] if args.type else adapter.get_available_data_types() + print(f"Синхронизация {len(data_types)} типов данных...") + print() + for data_type in data_types: + try: + result = await sync_service.sync_data_type( + data_type=data_type, + start_date=args.start_date, + end_date=args.end_date, + ) + print(f"✅ {data_type}: {result['added']} добавлено, {result['updated']} обновлено") + except Exception as e: + print(f"❌ {data_type}: {e}") + + print() + print("Синхронизация завершена!") + + +def cmd_sync(args): + asyncio.run(cmd_sync_async(args)) + + +def main(): + parser = argparse.ArgumentParser(prog=PROGRAM_NAME, description="MCP server for Mi Fitness data") + subparsers = parser.add_subparsers(dest="command", help="Available commands") + subparsers.add_parser("serve", help="Run MCP server") + + setup_parser = subparsers.add_parser("setup", help="Configure the server") + setup_parser.add_argument("--mode", choices=["mi_fitness_cloud"], help="Setup mode") + setup_parser.add_argument("--user-id", help="Mi Fitness user ID") + setup_parser.add_argument("--pass-token", help="Mi Fitness passToken") + setup_parser.add_argument("--region", help="Cloud region") + + subparsers.add_parser("doctor", help="Check configuration and diagnose issues") + + sync_parser = subparsers.add_parser("sync", help="Sync data from source") + sync_parser.add_argument("--type", choices=["daily_activity", "body_measurements", "heart_rate"], help="Type of data to sync") + sync_parser.add_argument("--start-date", help="Start date (YYYY-MM-DD)") + sync_parser.add_argument("--end-date", help="End date (YYYY-MM-DD)") + + args = parser.parse_args() + if args.command == "serve" or args.command is None: + asyncio.run(server_main()) + elif args.command == "setup": + cmd_setup(args) + elif args.command == "doctor": + cmd_doctor(args) + elif args.command == "sync": + cmd_sync(args) + else: + parser.print_help() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/mi_fitness_mcp/models/__init__.py b/src/mi_fitness_mcp/models/__init__.py new file mode 100644 index 0000000..f923af3 --- /dev/null +++ b/src/mi_fitness_mcp/models/__init__.py @@ -0,0 +1,183 @@ +"""Pydantic models for Mi Fitness MCP.""" + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class BaseEntity(BaseModel): + """Base entity with common fields.""" + + id: str = Field(description="Internal UUID") + provider: Literal["mi_fitness"] = Field(description="Data provider") + source_type: Literal["cloud_session"] = Field( + description="Source of the data" + ) + source_record_id: str | None = Field(None, description="Provider's native ID") + user_id: str = Field(description="User identifier") + device_id: str | None = Field(None, description="Device that recorded the data") + timezone: str = Field(default="UTC", description="Timezone for the data") + collected_at: datetime | None = Field(None, description="When data was recorded") + created_at: datetime = Field(default_factory=datetime.utcnow) + updated_at: datetime = Field(default_factory=datetime.utcnow) + + +class DailyActivity(BaseEntity): + """Daily activity summary (steps, calories, etc.).""" + + date: str = Field(description="Date in YYYY-MM-DD format") + steps: int = Field(ge=0, description="Number of steps") + distance_m: float = Field(ge=0, description="Distance in meters") + active_kcal: float = Field(ge=0, description="Active calories burned") + total_kcal: float | None = Field(None, ge=0, description="Total calories") + floors: int | None = Field(None, ge=0, description="Floors climbed") + active_minutes: int | None = Field(None, ge=0, description="Active minutes") + + +class SleepStage(BaseModel): + """Sleep stage information.""" + + stage: Literal["deep", "light", "rem", "awake"] = Field(description="Sleep stage") + minutes: int = Field(ge=0, description="Duration in minutes") + + +class SleepSession(BaseEntity): + """Sleep session data.""" + + sleep_id: str = Field(description="Unique sleep session ID") + start_at: datetime = Field(description="Sleep start time") + end_at: datetime = Field(description="Sleep end time") + duration_minutes: int = Field(ge=0, description="Total duration") + time_asleep_minutes: int = Field(ge=0, description="Actual sleep time") + time_awake_minutes: int = Field(ge=0, description="Time awake during sleep") + sleep_score: int | None = Field(None, ge=0, le=100, description="Sleep quality score") + is_nap: bool = Field(default=False, description="Whether this is a nap") + stages: list[SleepStage] = Field(default_factory=list, description="Sleep stages breakdown") + + +class Workout(BaseEntity): + """Workout/exercise session.""" + + workout_id: str = Field(description="Unique workout ID") + activity_type: str = Field(description="Type of activity (running, cycling, etc.)") + start_at: datetime = Field(description="Workout start time") + end_at: datetime = Field(description="Workout end time") + duration_minutes: int = Field(ge=0, description="Duration in minutes") + distance_m: float | None = Field(None, ge=0, description="Distance in meters") + calories_kcal: float | None = Field(None, ge=0, description="Calories burned") + avg_heart_rate_bpm: int | None = Field(None, ge=0, description="Average heart rate") + max_heart_rate_bpm: int | None = Field(None, ge=0, description="Maximum heart rate") + avg_pace_sec_per_km: float | None = Field(None, ge=0, description="Average pace") + max_pace_sec_per_km: float | None = Field(None, ge=0, description="Maximum pace") + total_steps: int | None = Field(None, ge=0, description="Steps during workout") + + +class BodyMeasurement(BaseEntity): + """Body composition measurement (from smart scale).""" + + timestamp: datetime = Field(description="Measurement time") + weight_kg: float = Field(gt=0, description="Weight in kilograms") + bmi: float | None = Field(None, gt=0, description="Body mass index") + body_fat_pct: float | None = Field(None, ge=0, le=100, description="Body fat percentage") + muscle_mass_kg: float | None = Field(None, ge=0, description="Muscle mass") + water_pct: float | None = Field(None, ge=0, le=100, description="Body water percentage") + bone_mass_kg: float | None = Field(None, ge=0, description="Bone mass") + visceral_fat_score: int | None = Field(None, ge=0, description="Visceral fat level") + basal_metabolism_kcal: int | None = Field(None, ge=0, description="BMR in kcal") + metabolic_age: int | None = Field(None, ge=0, description="Metabolic age") + + +class HeartRateSample(BaseEntity): + """Heart rate measurement.""" + + timestamp: datetime = Field(description="Measurement time") + bpm: int = Field(ge=0, description="Heart rate in beats per minute") + sample_type: Literal["resting", "active", "passive", "workout"] = Field( + default="passive", description="Type of measurement" + ) + + +class SpO2Sample(BaseEntity): + """Blood oxygen saturation measurement.""" + + timestamp: datetime = Field(description="Measurement time") + spo2_pct: int = Field(ge=0, le=100, description="SpO2 percentage") + + +class StressSample(BaseEntity): + """Stress level measurement.""" + + timestamp: datetime = Field(description="Measurement time") + stress_score: int = Field(ge=0, le=100, description="Stress score") + level: Literal["low", "medium", "high"] = Field(description="Stress level category") + + +class UserProfile(BaseModel): + """User profile information.""" + + user_id: str = Field(description="User identifier") + display_name: str | None = Field(None, description="Display name") + birth_year: int | None = Field(None, description="Birth year") + sex: Literal["male", "female", "other"] | None = Field(None, description="Sex") + height_cm: float | None = Field(None, gt=0, description="Height in cm") + timezone: str = Field(default="UTC", description="User timezone") + devices: list[dict[str, Any]] = Field(default_factory=list, description="Connected devices") + + +class DeviceInfo(BaseModel): + """Connected device information.""" + + device_id: str = Field(description="Device identifier") + device_type: str = Field(description="Type (band, watch, scale)") + model: str = Field(description="Device model") + firmware_version: str | None = Field(None, description="Firmware version") + last_sync_at: datetime | None = Field(None, description="Last sync time") + + +class DataCoverage(BaseModel): + """Data coverage information for a metric.""" + + data_type: str = Field(description="Type of data") + first_date: str | None = Field(None, description="First date with data") + last_date: str | None = Field(None, description="Last date with data") + days_with_data: int = Field(ge=0, description="Number of days with data") + gaps_detected: list[tuple[str, str]] = Field( + default_factory=list, description="Detected date gaps" + ) + + +class ConnectionStatus(BaseModel): + """Connection status response.""" + + mode: Literal["mi_fitness_cloud", "not_configured"] + connected: bool + last_sync_at: datetime | None = None + available_data_types: list[str] = Field(default_factory=list) + account_id_masked: str | None = None + sync_health: Literal["healthy", "stale", "error", "unknown"] = "unknown" + next_action: str | None = None + message: str | None = None + + +class SyncResult(BaseModel): + """Data synchronization result.""" + + sync_id: str = Field(description="Unique sync identifier") + started_at: datetime = Field(description="Sync start time") + finished_at: datetime = Field(description="Sync end time") + records_added: int = Field(ge=0, description="New records added") + records_updated: int = Field(ge=0, description="Records updated") + records_skipped: int = Field(ge=0, description="Records skipped (duplicates)") + data_types_synced: list[str] = Field(default_factory=list) + + +class QueryResponse(BaseModel): + """Generic query response envelope.""" + + status: Literal["ok", "error"] = "ok" + source: Literal["mi_fitness_cloud", "cloud_session", "cache", "unknown"] + generated_at: datetime = Field(default_factory=datetime.utcnow) + timezone: str = "UTC" + data: dict[str, Any] = Field(default_factory=dict) + error: str | None = None diff --git a/src/mi_fitness_mcp/server.py b/src/mi_fitness_mcp/server.py new file mode 100644 index 0000000..b0fb8e4 --- /dev/null +++ b/src/mi_fitness_mcp/server.py @@ -0,0 +1,181 @@ +"""MCP server implementation for Mi Fitness.""" + +import json +import logging +import uuid +from datetime import datetime +from typing import Any + +from mcp.server import Server +from mcp.server.stdio import stdio_server +from mcp.types import TextContent, Tool + +from mi_fitness_mcp.adapters.mi_fitness_cloud import MiFitnessCloudAdapter +from mi_fitness_mcp.auth import load_mi_fitness_token +from mi_fitness_mcp.config import load_config +from mi_fitness_mcp.models import ConnectionStatus, QueryResponse +from mi_fitness_mcp.services.query_service import QueryService +from mi_fitness_mcp.services.sync_service import SyncService +from mi_fitness_mcp.storage import Database + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +app = Server("mi-fitness-mcp") + +config = None +db = None +adapter = None +sync_service = None +query_service = None + + +@app.list_tools() +async def list_tools() -> list[Tool]: + return [ + Tool(name="get_connection_status", description="Check connection status", inputSchema={"type": "object", "properties": {}}), + Tool(name="sync_data", description="Synchronize Mi Fitness data", inputSchema={"type": "object", "properties": {"data_types": {"type": "array", "items": {"type": "string"}}, "start_date": {"type": "string"}, "end_date": {"type": "string"}, "force_full_sync": {"type": "boolean"}}}), + Tool(name="get_profile", description="Get user profile information", inputSchema={"type": "object", "properties": {}}), + Tool(name="get_daily_summary", description="Get daily activity summary", inputSchema={"type": "object", "properties": {"date": {"type": "string"}, "start_date": {"type": "string"}, "end_date": {"type": "string"}}}), + Tool(name="query_metric_series", description="Query metric series", inputSchema={"type": "object", "properties": {"metric": {"type": "string", "enum": ["steps", "distance_m", "active_kcal", "weight_kg"]}, "start_date": {"type": "string"}, "end_date": {"type": "string"}, "granularity": {"type": "string", "enum": ["day", "week", "month"]}, "aggregation": {"type": "string", "enum": ["sum", "avg", "min", "max", "latest"]}}, "required": ["metric", "start_date", "end_date"]}), + Tool(name="query_heart_rate", description="Query heart rate samples", inputSchema={"type": "object", "properties": {"start_date": {"type": "string"}, "end_date": {"type": "string"}, "sample_type": {"type": "string", "enum": ["resting", "active", "passive", "workout"]}, "limit": {"type": "integer"}}, "required": ["start_date", "end_date"]}), + Tool(name="query_body_measurements", description="Query body measurements", inputSchema={"type": "object", "properties": {"start_date": {"type": "string"}, "end_date": {"type": "string"}, "metrics": {"type": "array", "items": {"type": "string", "enum": ["weight_kg", "bmi", "body_fat_pct", "muscle_mass_kg", "water_pct"]}}, "latest_only": {"type": "boolean"}}, "required": ["start_date", "end_date"]}), + Tool(name="get_data_coverage", description="Get data coverage", inputSchema={"type": "object", "properties": {"data_types": {"type": "array", "items": {"type": "string"}}}}), + ] + + +@app.call_tool() +async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: + try: + if name == "get_connection_status": + result = await _handle_get_connection_status() + elif name == "sync_data": + result = await _handle_sync_data(arguments) + elif name == "get_profile": + result = await _handle_get_profile() + elif name == "get_daily_summary": + result = await _handle_get_daily_summary(arguments) + elif name == "query_metric_series": + result = await _handle_query_metric_series(arguments) + elif name == "query_heart_rate": + result = await _handle_query_heart_rate(arguments) + elif name == "query_body_measurements": + result = await _handle_query_body_measurements(arguments) + elif name == "get_data_coverage": + result = await _handle_get_data_coverage(arguments) + else: + result = {"status": "error", "error": f"Unknown tool: {name}"} + return [TextContent(type="text", text=json.dumps(result, default=str))] + except Exception as e: + logger.exception("Mi Fitness tool error") + return [TextContent(type="text", text=json.dumps({"status": "error", "error": str(e)}))] + + +async def _handle_get_connection_status() -> dict: + global adapter, config + if not config or config.mode == "not_configured": + return ConnectionStatus(mode="not_configured", connected=False, message="Server not configured.").model_dump() + + connected = adapter is not None and adapter.is_connected() + last_sync = None + available_types = [] + if db: + for data_type in ["daily_activity", "heart_rate", "body_measurements"]: + state = db.get_sync_state(data_type) + if state and state.get("last_sync_at"): + available_types.append(data_type) + sync_time = datetime.fromisoformat(state["last_sync_at"]) + if last_sync is None or sync_time > last_sync: + last_sync = sync_time + return ConnectionStatus(mode=config.mode, connected=connected, last_sync_at=last_sync, available_data_types=available_types).model_dump() + + +async def _handle_sync_data(arguments: dict) -> dict: + if not sync_service: + return {"status": "error", "error": "Sync service not initialized"} + data_types = arguments.get("data_types") or sync_service.adapter.get_available_data_types() + sync_id = str(uuid.uuid4()) + started_at = datetime.utcnow() + total_added = 0 + total_updated = 0 + total_skipped = 0 + types_synced = [] + for data_type in data_types: + try: + result = await sync_service.sync_data_type(data_type=data_type, start_date=arguments.get("start_date"), end_date=arguments.get("end_date"), force_full=arguments.get("force_full_sync", False)) + total_added += result.get("added", 0) + total_updated += result.get("updated", 0) + total_skipped += result.get("skipped", 0) + types_synced.append(data_type) + except Exception as e: + logger.error(f"Failed to sync {data_type}: {e}") + finished_at = datetime.utcnow() + return {"status": "ok", "sync_id": sync_id, "started_at": started_at.isoformat(), "finished_at": finished_at.isoformat(), "records_added": total_added, "records_updated": total_updated, "records_skipped": total_skipped, "data_types_synced": types_synced} + + +async def _handle_get_profile() -> dict: + if not adapter or not adapter.is_connected(): + return {"status": "error", "error": "Not connected to data source"} + return QueryResponse(status="ok", source=config.mode if config else "unknown", data={"profile": {"user_id": adapter.get_user_id() or "unknown", "timezone": config.timezone if config else "UTC", "devices": []}}).model_dump() + + +async def _handle_get_daily_summary(arguments: dict) -> dict: + if not query_service: + return {"status": "error", "error": "Query service not initialized"} + start_date = arguments.get("date") or arguments.get("start_date") + end_date = arguments.get("date") or arguments.get("end_date") + if not start_date or not end_date: + return {"status": "error", "error": "date or start_date/end_date required"} + summaries = query_service.get_daily_summaries(start_date, end_date) + return QueryResponse(status="ok", source="cache", data={"summaries": summaries}).model_dump() + + +async def _handle_query_metric_series(arguments: dict) -> dict: + if not query_service: + return {"status": "error", "error": "Query service not initialized"} + series = query_service.get_metric_series(metric=arguments["metric"], start_date=arguments["start_date"], end_date=arguments["end_date"], granularity=arguments.get("granularity", "day"), aggregation=arguments.get("aggregation", "sum")) + return QueryResponse(status="ok", source="cache", data={"metric": arguments["metric"], "series": series}).model_dump() + + +async def _handle_query_heart_rate(arguments: dict) -> dict: + if not query_service: + return {"status": "error", "error": "Query service not initialized"} + samples = query_service.get_heart_rate_samples(start_date=arguments["start_date"], end_date=arguments["end_date"], sample_type=arguments.get("sample_type"), limit=arguments.get("limit")) + return QueryResponse(status="ok", source="cache", data={"samples": samples, "count": len(samples)}).model_dump() + + +async def _handle_query_body_measurements(arguments: dict) -> dict: + if not query_service: + return {"status": "error", "error": "Query service not initialized"} + measurements = query_service.get_body_measurements(start_date=arguments["start_date"], end_date=arguments["end_date"], metrics=arguments.get("metrics")) + if arguments.get("latest_only") and measurements: + measurements = [measurements[-1]] + return QueryResponse(status="ok", source="cache", data={"measurements": measurements, "count": len(measurements)}).model_dump() + + +async def _handle_get_data_coverage(arguments: dict) -> dict: + if not query_service: + return {"status": "error", "error": "Query service not initialized"} + coverage = query_service.get_data_coverage(arguments.get("data_types")) + return QueryResponse(status="ok", source="cache", data={"coverage": coverage}).model_dump() + + +async def main(): + global config, db, adapter, sync_service, query_service + config = load_config() + db = Database(config.database_path) + if config.mode == "mi_fitness_cloud": + user_id, pass_token = load_mi_fitness_token() + if user_id and pass_token: + adapter = MiFitnessCloudAdapter(user_id=user_id, pass_token=pass_token, region=config.region) + if await adapter.connect(): + logger.info("Connected to Mi Fitness cloud API") + else: + logger.warning("Failed to connect to Mi Fitness cloud API") + if adapter: + sync_service = SyncService(adapter, db) + query_service = QueryService(db, adapter.get_user_id() or "unknown") + else: + query_service = QueryService(db, "unknown") + async with stdio_server() as (read_stream, write_stream): + await app.run(read_stream, write_stream, app.create_initialization_options()) diff --git a/src/mi_fitness_mcp/services/__init__.py b/src/mi_fitness_mcp/services/__init__.py new file mode 100644 index 0000000..e6136ac --- /dev/null +++ b/src/mi_fitness_mcp/services/__init__.py @@ -0,0 +1,6 @@ +"""Service exports.""" + +from mi_fitness_mcp.services.query_service import QueryService +from mi_fitness_mcp.services.sync_service import SyncService + +__all__ = ["QueryService", "SyncService"] diff --git a/src/mi_fitness_mcp/services/autosync.py b/src/mi_fitness_mcp/services/autosync.py new file mode 100644 index 0000000..4ad656e --- /dev/null +++ b/src/mi_fitness_mcp/services/autosync.py @@ -0,0 +1,223 @@ +"""Background sync scheduler for Mi Fitness MCP.""" + +import asyncio +import logging +from datetime import datetime, timedelta +from typing import Callable + +from mi_fitness_mcp.adapters.base import DataAdapter +from mi_fitness_mcp.storage import Database + +logger = logging.getLogger(__name__) + + +class SyncScheduler: + """Scheduler for automatic background synchronization.""" + + def __init__( + self, + adapter: DataAdapter, + db: Database, + sync_interval_minutes: int = 60, + ): + """Initialize sync scheduler. + + Args: + adapter: Data source adapter + db: Database instance + sync_interval_minutes: How often to sync (default: 1 hour) + """ + self.adapter = adapter + self.db = db + self.sync_interval = timedelta(minutes=sync_interval_minutes) + self._running = False + self._task: asyncio.Task | None = None + self._last_sync: datetime | None = None + self._sync_callback: Callable | None = None + + def set_sync_callback(self, callback: Callable) -> None: + """Set callback to be called after each sync. + + Args: + callback: Function to call with sync results + """ + self._sync_callback = callback + + async def start(self) -> None: + """Start the sync scheduler.""" + if self._running: + logger.warning("Scheduler already running") + return + + self._running = True + self._task = asyncio.create_task(self._run_scheduler()) + logger.info(f"Sync scheduler started (interval: {self.sync_interval})") + + async def stop(self) -> None: + """Stop the sync scheduler.""" + if not self._running: + return + + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + logger.info("Sync scheduler stopped") + + async def _run_scheduler(self) -> None: + """Main scheduler loop.""" + from mi_fitness_mcp.services.sync_service import SyncService + + sync_service = SyncService(self.adapter, self.db) + + while self._running: + try: + # Check if it's time to sync + if self._should_sync(): + await self._perform_sync(sync_service) + + # Wait for next check + await asyncio.sleep(60) # Check every minute + + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"Error in sync scheduler: {e}") + await asyncio.sleep(60) # Wait before retrying + + def _should_sync(self) -> bool: + """Check if sync should be performed.""" + if not self._last_sync: + return True + + time_since_last = datetime.utcnow() - self._last_sync + return time_since_last >= self.sync_interval + + async def _perform_sync(self, sync_service) -> None: + """Perform synchronization of all data types.""" + logger.info("Starting automatic sync") + + results = {} + data_types = self.adapter.get_available_data_types() + + for data_type in data_types: + try: + result = sync_service.sync_data_type(data_type) + results[data_type] = result + logger.info( + f"Synced {data_type}: {result['added']} added, " + f"{result['updated']} updated" + ) + except Exception as e: + logger.error(f"Failed to sync {data_type}: {e}") + results[data_type] = {"error": str(e)} + + self._last_sync = datetime.utcnow() + + # Call callback if set + if self._sync_callback: + try: + self._sync_callback(results) + except Exception as e: + logger.error(f"Sync callback failed: {e}") + + logger.info("Automatic sync completed") + + def get_status(self) -> dict: + """Get scheduler status.""" + return { + "running": self._running, + "last_sync": self._last_sync.isoformat() if self._last_sync else None, + "sync_interval_minutes": self.sync_interval.total_seconds() / 60, + "next_sync": ( + (self._last_sync + self.sync_interval).isoformat() + if self._last_sync + else None + ), + } + + +class AutoSyncManager: + """Manager for automatic synchronization configuration.""" + + def __init__(self, config_path: str = "config/autosync.json"): + """Initialize auto-sync manager. + + Args: + config_path: Path to autosync configuration file + """ + self.config_path = config_path + self._scheduler: SyncScheduler | None = None + + def load_config(self) -> dict: + """Load auto-sync configuration.""" + import json + from pathlib import Path + + path = Path(self.config_path) + if not path.exists(): + return { + "enabled": False, + "sync_interval_minutes": 60, + "data_types": ["daily_activity", "sleep", "workouts"], + } + + with open(path, "r") as f: + return json.load(f) + + def save_config(self, config: dict) -> None: + """Save auto-sync configuration.""" + import json + from pathlib import Path + + path = Path(self.config_path) + path.parent.mkdir(parents=True, exist_ok=True) + + with open(path, "w") as f: + json.dump(config, f, indent=2) + + async def start_scheduler( + self, + adapter: DataAdapter, + db: Database, + ) -> SyncScheduler | None: + """Start scheduler with loaded configuration. + + Args: + adapter: Data source adapter + db: Database instance + + Returns: + SyncScheduler instance or None if not enabled + """ + config = self.load_config() + + if not config.get("enabled", False): + logger.info("Auto-sync is disabled") + return None + + interval = config.get("sync_interval_minutes", 60) + + self._scheduler = SyncScheduler( + adapter=adapter, + db=db, + sync_interval_minutes=interval, + ) + + await self._scheduler.start() + return self._scheduler + + async def stop_scheduler(self) -> None: + """Stop the running scheduler.""" + if self._scheduler: + await self._scheduler.stop() + self._scheduler = None + + def get_scheduler_status(self) -> dict | None: + """Get current scheduler status.""" + if not self._scheduler: + return None + return self._scheduler.get_status() diff --git a/src/mi_fitness_mcp/services/query_service.py b/src/mi_fitness_mcp/services/query_service.py new file mode 100644 index 0000000..564b0a3 --- /dev/null +++ b/src/mi_fitness_mcp/services/query_service.py @@ -0,0 +1,301 @@ +"""Query service for retrieving data from database.""" + +from datetime import datetime, timedelta +from typing import Any + +from mi_fitness_mcp.storage import Database + + +class QueryService: + """Service for querying fitness data from local database.""" + + def __init__(self, db: Database, user_id: str): + """Initialize query service. + + Args: + db: Database instance + user_id: User identifier + """ + self.db = db + self.user_id = user_id + + def get_daily_summaries(self, start_date: str, end_date: str) -> list[dict[str, Any]]: + """Get daily activity summaries for date range.""" + records = self.db.query_daily_activity(self.user_id, start_date, end_date) + + # Group by date and aggregate + summaries = {} + for record in records: + date = record["date"] + if date not in summaries: + summaries[date] = { + "date": date, + "steps": 0, + "distance_m": 0, + "active_kcal": 0, + "total_kcal": 0, + "floors": 0, + "active_minutes": 0, + } + + summaries[date]["steps"] += record.get("steps", 0) + summaries[date]["distance_m"] += record.get("distance_m", 0) + summaries[date]["active_kcal"] += record.get("active_kcal", 0) + if record.get("total_kcal"): + summaries[date]["total_kcal"] += record["total_kcal"] + if record.get("floors"): + summaries[date]["floors"] += record["floors"] + if record.get("active_minutes"): + summaries[date]["active_minutes"] += record["active_minutes"] + + return list(summaries.values()) + + def get_metric_series( + self, + metric: str, + start_date: str, + end_date: str, + granularity: str = "day", + aggregation: str = "sum", + ) -> list[dict[str, Any]]: + """Get time series for a metric.""" + summaries = self.get_daily_summaries(start_date, end_date) + + series = [] + for summary in summaries: + value = summary.get(metric) + if value is not None: + series.append({ + "date": summary["date"], + "value": value, + }) + + # Apply aggregation if needed + if granularity == "week": + series = self._aggregate_by_week(series, aggregation) + elif granularity == "month": + series = self._aggregate_by_month(series, aggregation) + + return series + + def _aggregate_by_week( + self, + series: list[dict], + aggregation: str, + ) -> list[dict]: + """Aggregate daily series by week.""" + weeks = {} + + for item in series: + date = datetime.strptime(item["date"], "%Y-%m-%d") + # Get week start (Monday) + week_start = date - timedelta(days=date.weekday()) + week_key = week_start.strftime("%Y-%m-%d") + + if week_key not in weeks: + weeks[week_key] = [] + weeks[week_key].append(item["value"]) + + result = [] + for week_key, values in sorted(weeks.items()): + if aggregation == "sum": + value = sum(values) + elif aggregation == "avg": + value = sum(values) / len(values) + elif aggregation == "min": + value = min(values) + elif aggregation == "max": + value = max(values) + else: + value = sum(values) + + result.append({"date": week_key, "value": value}) + + return result + + def _aggregate_by_month( + self, + series: list[dict], + aggregation: str, + ) -> list[dict]: + """Aggregate daily series by month.""" + months = {} + + for item in series: + date = datetime.strptime(item["date"], "%Y-%m-%d") + month_key = date.strftime("%Y-%m") + + if month_key not in months: + months[month_key] = [] + months[month_key].append(item["value"]) + + result = [] + for month_key, values in sorted(months.items()): + if aggregation == "sum": + value = sum(values) + elif aggregation == "avg": + value = sum(values) / len(values) + elif aggregation == "min": + value = min(values) + elif aggregation == "max": + value = max(values) + else: + value = sum(values) + + result.append({"date": month_key + "-01", "value": value}) + + return result + + def get_sleep_sessions( + self, + start_date: str, + end_date: str, + include_naps: bool = True, + ) -> list[dict[str, Any]]: + """Get sleep sessions for date range.""" + records = self.db.query_sleep_sessions(self.user_id, start_date, end_date) + + sessions = [] + for record in records: + # Skip naps if not included + if not include_naps and record.get("is_nap"): + continue + + session = { + "sleep_id": record["sleep_id"], + "start_at": record["start_at"], + "end_at": record["end_at"], + "duration_minutes": record["duration_minutes"], + "time_asleep_minutes": record["time_asleep_minutes"], + "time_awake_minutes": record["time_awake_minutes"], + "sleep_score": record.get("sleep_score"), + "is_nap": record.get("is_nap", False), + } + + # Parse stages if available + if record.get("stages"): + import json + try: + session["stages"] = json.loads(record["stages"]) + except json.JSONDecodeError: + pass + + sessions.append(session) + + return sessions + + def get_workouts( + self, + start_date: str, + end_date: str, + activity_types: list[str] | None = None, + min_duration: int | None = None, + min_distance_km: float | None = None, + ) -> list[dict[str, Any]]: + """Get workouts for date range with optional filters.""" + records = self.db.query_workouts(self.user_id, start_date, end_date) + + workouts = [] + for record in records: + # Apply filters + if activity_types: + if record["activity_type"].lower() not in [t.lower() for t in activity_types]: + continue + + if min_duration and record.get("duration_minutes", 0) < min_duration: + continue + + if min_distance_km: + distance_km = (record.get("distance_m") or 0) / 1000 + if distance_km < min_distance_km: + continue + + workouts.append({ + "workout_id": record["workout_id"], + "activity_type": record["activity_type"], + "start_at": record["start_at"], + "end_at": record["end_at"], + "duration_minutes": record["duration_minutes"], + "distance_m": record.get("distance_m"), + "calories_kcal": record.get("calories_kcal"), + "avg_heart_rate_bpm": record.get("avg_heart_rate_bpm"), + "max_heart_rate_bpm": record.get("max_heart_rate_bpm"), + "avg_pace_sec_per_km": record.get("avg_pace_sec_per_km"), + "max_pace_sec_per_km": record.get("max_pace_sec_per_km"), + "total_steps": record.get("total_steps"), + }) + + return workouts + + def get_body_measurements( + self, + start_date: str, + end_date: str, + metrics: list[str] | None = None, + ) -> list[dict[str, Any]]: + """Get body measurements for date range.""" + records = self.db.query_body_measurements(self.user_id, start_date, end_date) + + measurements = [] + for record in records: + measurement = { + "timestamp": record["timestamp"], + "weight_kg": record["weight_kg"], + } + + # Add optional metrics + optional_fields = [ + "bmi", "body_fat_pct", "muscle_mass_kg", "water_pct", + "bone_mass_kg", "visceral_fat_score", "basal_metabolism_kcal", + "metabolic_age", + ] + + for field in optional_fields: + if record.get(field) is not None: + measurement[field] = record[field] + + # Filter metrics if specified + if metrics: + filtered = {"timestamp": measurement["timestamp"]} + for metric in metrics: + if metric in measurement: + filtered[metric] = measurement[metric] + measurement = filtered + + measurements.append(measurement) + + return measurements + + def get_heart_rate_samples( + self, + start_date: str, + end_date: str, + sample_type: str | None = None, + limit: int | None = None, + ) -> list[dict[str, Any]]: + records = self.db.query_heart_rate_samples(self.user_id, start_date, end_date) + + samples = [] + for record in records: + if sample_type and record.get("sample_type") != sample_type: + continue + samples.append( + { + "timestamp": record["timestamp"], + "bpm": record["bpm"], + "sample_type": record.get("sample_type", "passive"), + } + ) + + if limit is not None: + return samples[:limit] + return samples + + 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) + + if data_types: + coverage = [c for c in coverage if c["data_type"] in data_types] + + return coverage diff --git a/src/mi_fitness_mcp/services/sync_service.py b/src/mi_fitness_mcp/services/sync_service.py new file mode 100644 index 0000000..ed7f31c --- /dev/null +++ b/src/mi_fitness_mcp/services/sync_service.py @@ -0,0 +1,164 @@ +"""Sync service for importing data from adapters to database.""" + +import asyncio +import logging +from collections.abc import AsyncIterator, Iterator +from datetime import datetime, timedelta +from typing import Any + +from mi_fitness_mcp.adapters.base import DataAdapter +from mi_fitness_mcp.storage import Database + +logger = logging.getLogger(__name__) + + +class SyncService: + """Service for synchronizing data from adapters to local database.""" + + def __init__(self, adapter: DataAdapter, db: Database): + """Initialize sync service. + + Args: + adapter: Data source adapter + db: Database instance + """ + self.adapter = adapter + self.db = db + + async def _iterate_records(self, records: Any) -> AsyncIterator[Any]: + if hasattr(records, "__aiter__"): + async for record in records: + yield record + return + + for record in records: + yield record + + async def sync_data_type( + self, + data_type: str, + start_date: str | None = None, + end_date: str | None = None, + force_full: bool = False, + ) -> dict: + """Synchronize a specific data type. + + Args: + data_type: Type of data to sync (daily_activity, sleep, workouts, body_measurements) + start_date: Start date (YYYY-MM-DD), defaults to 30 days ago + end_date: End date (YYYY-MM-DD), defaults to today + force_full: Force full sync ignoring last sync state + + Returns: + Dict with sync statistics + """ + if not self.adapter.is_connected(): + raise RuntimeError("Adapter not connected") + + # Get last sync state for incremental sync + last_record_ts = None + if not force_full: + state = self.db.get_sync_state(data_type) + if state and state.get("last_record_timestamp"): + last_record_ts = datetime.fromisoformat(state["last_record_timestamp"]) + + if not end_date: + end_date = datetime.now().strftime("%Y-%m-%d") + if not start_date: + if last_record_ts: + start_date = last_record_ts.strftime("%Y-%m-%d") + elif self.adapter.__class__.__name__ == "MiFitnessCloudAdapter": + start_date = "2020-01-01" + else: + start = datetime.strptime(end_date, "%Y-%m-%d") - timedelta(days=30) + start_date = start.strftime("%Y-%m-%d") + + added = 0 + updated = 0 + skipped = 0 + last_ts = None + + # Sync based on data type + if data_type == "daily_activity": + records = self.adapter.iter_daily_activity(start_date, end_date) + async for activity in self._iterate_records(records): + if self.db.insert_daily_activity(activity): + added += 1 + else: + updated += 1 + if last_ts is None or (activity.collected_at and activity.collected_at > last_ts): + last_ts = activity.collected_at + + elif data_type == "sleep": + records = self.adapter.iter_sleep_sessions(start_date, end_date) + async for sleep in self._iterate_records(records): + if self.db.insert_sleep_session(sleep): + added += 1 + else: + updated += 1 + if last_ts is None or sleep.start_at > last_ts: + last_ts = sleep.start_at + + elif data_type == "workouts": + records = self.adapter.iter_workouts(start_date, end_date) + async for workout in self._iterate_records(records): + if self.db.insert_workout(workout): + added += 1 + else: + updated += 1 + if last_ts is None or workout.start_at > last_ts: + last_ts = workout.start_at + + elif data_type == "body_measurements": + records = self.adapter.iter_body_measurements(start_date, end_date) + async for measurement in self._iterate_records(records): + if self.db.insert_body_measurement(measurement): + added += 1 + else: + updated += 1 + if last_ts is None or measurement.timestamp > last_ts: + last_ts = measurement.timestamp + + elif data_type == "heart_rate": + records = self.adapter.iter_heart_rate(start_date, end_date) + async for sample in self._iterate_records(records): + if self.db.insert_heart_rate_sample(sample): + added += 1 + else: + updated += 1 + if last_ts is None or sample.timestamp > last_ts: + last_ts = sample.timestamp + + else: + raise ValueError(f"Unknown data type: {data_type}") + + # Update sync state + if last_ts: + self.db.update_sync_state(data_type, last_ts) + + logger.info( + f"Synced {data_type}: {added} added, {updated} updated, " + f"range {start_date} to {end_date}" + ) + + return { + "data_type": data_type, + "added": added, + "updated": updated, + "skipped": skipped, + "start_date": start_date, + "end_date": end_date, + } + + def sync_data_type_sync( + self, + data_type: str, + start_date: str | None = None, + end_date: str | None = None, + force_full: bool = False, + ) -> dict: + """Synchronous wrapper for sync_data_type. + + Use this when calling from synchronous code. + """ + return asyncio.run(self.sync_data_type(data_type, start_date, end_date, force_full)) diff --git a/src/mi_fitness_mcp/storage/__init__.py b/src/mi_fitness_mcp/storage/__init__.py new file mode 100644 index 0000000..2cbcd41 --- /dev/null +++ b/src/mi_fitness_mcp/storage/__init__.py @@ -0,0 +1,619 @@ +"""SQLite storage layer for Mi Fitness MCP.""" + +import json +import sqlite3 +from contextlib import contextmanager +from datetime import datetime +from pathlib import Path +from typing import Any + +from mi_fitness_mcp.models import ( + BodyMeasurement, + DailyActivity, + HeartRateSample, + SleepSession, + Workout, +) + + +class Database: + """SQLite database manager.""" + + def __init__(self, db_path: Path | str): + """Initialize database. + + Args: + db_path: Path to SQLite database file + """ + self.db_path = Path(db_path) + self._init_db() + + def _init_db(self) -> None: + """Initialize database schema.""" + self.db_path.parent.mkdir(parents=True, exist_ok=True) + + with self._get_connection() as conn: + # Daily activity table + conn.execute(""" + CREATE TABLE IF NOT EXISTS daily_activity ( + 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, + date TEXT NOT NULL, + steps INTEGER NOT NULL DEFAULT 0, + distance_m REAL NOT NULL DEFAULT 0, + active_kcal REAL NOT NULL DEFAULT 0, + total_kcal REAL, + floors INTEGER, + active_minutes INTEGER, + UNIQUE(user_id, date, device_id) + ) + """) + + # Sleep sessions table + conn.execute(""" + CREATE TABLE IF NOT EXISTS sleep_sessions ( + 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, + sleep_id TEXT NOT NULL, + start_at TIMESTAMP NOT NULL, + end_at TIMESTAMP NOT NULL, + duration_minutes INTEGER NOT NULL, + time_asleep_minutes INTEGER NOT NULL, + time_awake_minutes INTEGER NOT NULL, + sleep_score INTEGER, + is_nap BOOLEAN DEFAULT FALSE, + stages TEXT, -- JSON array of sleep stages + UNIQUE(user_id, sleep_id) + ) + """) + + # Workouts table + conn.execute(""" + CREATE TABLE IF NOT EXISTS workouts ( + 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, + workout_id TEXT NOT NULL, + activity_type TEXT NOT NULL, + start_at TIMESTAMP NOT NULL, + end_at TIMESTAMP NOT NULL, + duration_minutes INTEGER NOT NULL, + distance_m REAL, + calories_kcal REAL, + avg_heart_rate_bpm INTEGER, + max_heart_rate_bpm INTEGER, + avg_pace_sec_per_km REAL, + max_pace_sec_per_km REAL, + total_steps INTEGER, + UNIQUE(user_id, workout_id) + ) + """) + + # Body measurements table + conn.execute(""" + CREATE TABLE IF NOT EXISTS body_measurements ( + 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, + weight_kg REAL NOT NULL, + bmi REAL, + body_fat_pct REAL, + muscle_mass_kg REAL, + water_pct REAL, + bone_mass_kg REAL, + visceral_fat_score INTEGER, + basal_metabolism_kcal INTEGER, + metabolic_age INTEGER, + UNIQUE(user_id, timestamp, device_id) + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS heart_rate_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, + bpm INTEGER NOT NULL, + sample_type TEXT NOT NULL, + UNIQUE(user_id, timestamp, sample_type) + ) + """) + + # Sync state table + conn.execute(""" + CREATE TABLE IF NOT EXISTS sync_state ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + data_type TEXT NOT NULL UNIQUE, + last_sync_at TIMESTAMP, + last_record_timestamp TIMESTAMP, + records_count INTEGER DEFAULT 0 + ) + """) + + # Create indexes for better query performance + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_activity_user_date + ON daily_activity(user_id, date) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_sleep_user_start + ON sleep_sessions(user_id, start_at) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_workouts_user_start + ON workouts(user_id, start_at) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_measurements_user_ts + ON body_measurements(user_id, timestamp) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_heart_rate_user_ts + ON heart_rate_samples(user_id, timestamp) + """) + + conn.commit() + + @contextmanager + def _get_connection(self): + """Get database connection with row factory.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + def insert_daily_activity(self, activity: DailyActivity) -> bool: + """Insert or update daily activity record. + + Returns: + True if record was inserted, False if updated + """ + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO daily_activity ( + id, provider, source_type, source_record_id, user_id, device_id, + timezone, collected_at, date, steps, distance_m, active_kcal, + total_kcal, floors, active_minutes + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + steps = excluded.steps, + distance_m = excluded.distance_m, + active_kcal = excluded.active_kcal, + total_kcal = excluded.total_kcal, + floors = excluded.floors, + active_minutes = excluded.active_minutes, + updated_at = CURRENT_TIMESTAMP + """, + ( + activity.id, + activity.provider, + activity.source_type, + activity.source_record_id, + activity.user_id, + activity.device_id, + activity.timezone, + activity.collected_at.isoformat() if activity.collected_at else None, + activity.date, + activity.steps, + activity.distance_m, + activity.active_kcal, + activity.total_kcal, + activity.floors, + activity.active_minutes, + ), + ) + conn.commit() + return cursor.rowcount > 0 + + def insert_sleep_session(self, sleep: SleepSession) -> bool: + """Insert or update sleep session record.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO sleep_sessions ( + id, provider, source_type, source_record_id, user_id, device_id, + timezone, collected_at, sleep_id, start_at, end_at, duration_minutes, + time_asleep_minutes, time_awake_minutes, sleep_score, is_nap, stages + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id, sleep_id) DO UPDATE SET + duration_minutes = excluded.duration_minutes, + time_asleep_minutes = excluded.time_asleep_minutes, + time_awake_minutes = excluded.time_awake_minutes, + sleep_score = excluded.sleep_score, + stages = excluded.stages, + updated_at = CURRENT_TIMESTAMP + """, + ( + sleep.id, + sleep.provider, + sleep.source_type, + sleep.source_record_id, + sleep.user_id, + sleep.device_id, + sleep.timezone, + sleep.collected_at.isoformat() if sleep.collected_at else None, + sleep.sleep_id, + sleep.start_at.isoformat(), + sleep.end_at.isoformat(), + sleep.duration_minutes, + sleep.time_asleep_minutes, + sleep.time_awake_minutes, + sleep.sleep_score, + sleep.is_nap, + json.dumps([s.model_dump() for s in sleep.stages]), + ), + ) + conn.commit() + return cursor.rowcount > 0 + + def insert_workout(self, workout: Workout) -> bool: + """Insert or update workout record.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO workouts ( + id, provider, source_type, source_record_id, user_id, device_id, + timezone, collected_at, workout_id, activity_type, start_at, end_at, + duration_minutes, distance_m, calories_kcal, avg_heart_rate_bpm, + max_heart_rate_bpm, avg_pace_sec_per_km, max_pace_sec_per_km, total_steps + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id, workout_id) DO UPDATE SET + duration_minutes = excluded.duration_minutes, + distance_m = excluded.distance_m, + calories_kcal = excluded.calories_kcal, + avg_heart_rate_bpm = excluded.avg_heart_rate_bpm, + max_heart_rate_bpm = excluded.max_heart_rate_bpm, + updated_at = CURRENT_TIMESTAMP + """, + ( + workout.id, + workout.provider, + workout.source_type, + workout.source_record_id, + workout.user_id, + workout.device_id, + workout.timezone, + workout.collected_at.isoformat() if workout.collected_at else None, + workout.workout_id, + workout.activity_type, + workout.start_at.isoformat(), + workout.end_at.isoformat(), + workout.duration_minutes, + workout.distance_m, + workout.calories_kcal, + workout.avg_heart_rate_bpm, + workout.max_heart_rate_bpm, + workout.avg_pace_sec_per_km, + workout.max_pace_sec_per_km, + workout.total_steps, + ), + ) + conn.commit() + return cursor.rowcount > 0 + + def insert_body_measurement(self, measurement: BodyMeasurement) -> bool: + """Insert or update body measurement record.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO body_measurements ( + id, provider, source_type, source_record_id, user_id, device_id, + timezone, collected_at, timestamp, weight_kg, bmi, body_fat_pct, + muscle_mass_kg, water_pct, bone_mass_kg, visceral_fat_score, + basal_metabolism_kcal, metabolic_age + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + weight_kg = excluded.weight_kg, + bmi = excluded.bmi, + body_fat_pct = excluded.body_fat_pct, + muscle_mass_kg = excluded.muscle_mass_kg, + water_pct = excluded.water_pct, + updated_at = CURRENT_TIMESTAMP + """, + ( + measurement.id, + measurement.provider, + measurement.source_type, + measurement.source_record_id, + measurement.user_id, + measurement.device_id, + measurement.timezone, + measurement.collected_at.isoformat() if measurement.collected_at else None, + measurement.timestamp.isoformat(), + measurement.weight_kg, + measurement.bmi, + measurement.body_fat_pct, + measurement.muscle_mass_kg, + measurement.water_pct, + measurement.bone_mass_kg, + measurement.visceral_fat_score, + measurement.basal_metabolism_kcal, + measurement.metabolic_age, + ), + ) + conn.commit() + return cursor.rowcount > 0 + + def insert_heart_rate_sample(self, sample: HeartRateSample) -> bool: + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO heart_rate_samples ( + id, provider, source_type, source_record_id, user_id, device_id, + timezone, collected_at, timestamp, bpm, sample_type + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + bpm = excluded.bpm, + 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.bpm, + sample.sample_type, + ), + ) + 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: + conn.execute( + """ + INSERT INTO sync_state (data_type, last_sync_at, last_record_timestamp) + VALUES (?, CURRENT_TIMESTAMP, ?) + ON CONFLICT(data_type) DO UPDATE SET + last_sync_at = CURRENT_TIMESTAMP, + last_record_timestamp = excluded.last_record_timestamp + """, + (data_type, last_record_ts.isoformat() if last_record_ts else None), + ) + conn.commit() + + def get_sync_state(self, data_type: str) -> dict[str, Any] | None: + """Get sync state for a data type.""" + with self._get_connection() as conn: + row = conn.execute( + "SELECT * FROM sync_state WHERE data_type = ?", + (data_type,), + ).fetchone() + return dict(row) if row else None + + def query_daily_activity( + self, + user_id: str, + start_date: str, + end_date: str, + ) -> list[dict[str, Any]]: + """Query daily activity records.""" + with self._get_connection() as conn: + rows = conn.execute( + """ + SELECT * FROM daily_activity + WHERE user_id = ? AND date >= ? AND date <= ? + ORDER BY date + """, + (user_id, start_date, end_date), + ).fetchall() + return [dict(row) for row in rows] + + def query_sleep_sessions( + self, + user_id: str, + start_date: str, + end_date: str, + ) -> list[dict[str, Any]]: + """Query sleep session records.""" + with self._get_connection() as conn: + rows = conn.execute( + """ + SELECT * FROM sleep_sessions + 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 query_workouts( + self, + user_id: str, + start_date: str, + end_date: str, + ) -> list[dict[str, Any]]: + """Query workout records.""" + with self._get_connection() as conn: + rows = conn.execute( + """ + SELECT * FROM workouts + 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 query_body_measurements( + self, + user_id: str, + start_date: str, + end_date: str, + ) -> list[dict[str, Any]]: + """Query body measurement records.""" + with self._get_connection() as conn: + rows = conn.execute( + """ + SELECT * FROM body_measurements + 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_heart_rate_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 heart_rate_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 get_data_coverage(self, user_id: str) -> list[dict[str, Any]]: + """Get data coverage statistics.""" + with self._get_connection() as conn: + results = [] + + # Activity coverage + row = conn.execute( + """ + SELECT + MIN(date) as first_date, + MAX(date) as last_date, + COUNT(DISTINCT date) as days_with_data + FROM daily_activity + WHERE user_id = ? + """, + (user_id,), + ).fetchone() + if row and row["first_date"]: + results.append({ + "data_type": "daily_activity", + **dict(row), + }) + + # Sleep coverage + row = conn.execute( + """ + SELECT + MIN(date(start_at)) as first_date, + MAX(date(start_at)) as last_date, + COUNT(DISTINCT date(start_at)) as days_with_data + FROM sleep_sessions + WHERE user_id = ? + """, + (user_id,), + ).fetchone() + if row and row["first_date"]: + results.append({ + "data_type": "sleep", + **dict(row), + }) + + # Workouts coverage + row = conn.execute( + """ + SELECT + MIN(date(start_at)) as first_date, + MAX(date(start_at)) as last_date, + COUNT(DISTINCT date(start_at)) as days_with_data + FROM workouts + WHERE user_id = ? + """, + (user_id,), + ).fetchone() + if row and row["first_date"]: + results.append({ + "data_type": "workouts", + **dict(row), + }) + + # Body measurements coverage + row = conn.execute( + """ + SELECT + MIN(date(timestamp)) as first_date, + MAX(date(timestamp)) as last_date, + COUNT(DISTINCT date(timestamp)) as days_with_data + FROM body_measurements + WHERE user_id = ? + """, + (user_id,), + ).fetchone() + if row and row["first_date"]: + results.append({ + "data_type": "body_measurements", + **dict(row), + }) + + row = conn.execute( + """ + SELECT + MIN(date(timestamp)) as first_date, + MAX(date(timestamp)) as last_date, + COUNT(DISTINCT date(timestamp)) as days_with_data + FROM heart_rate_samples + WHERE user_id = ? + """, + (user_id,), + ).fetchone() + if row and row["first_date"]: + results.append({ + "data_type": "heart_rate", + **dict(row), + }) + + return results