量产报告服务: 实现五个查询 RPC 的 Mongo 聚合与 typed 出参

.env.example

@@ -32,6 +32,12 @@ SOMNI_MONGO_URI=mongodb://user:password@host:27017/Somni
 SOMNI_MONGO_DB=Somni
 # 答卷集合（对齐 BioNode quiz_answers 结构）
 SOMNI_MONGO_ANSWERS_COLLECTION=somni_quiz_answers
+# 量产报告相关集合
+SOMNI_MONGO_DEVICES_COLLECTION=somni_devices
+SOMNI_MONGO_TELEMETRY_COLLECTION=somni_telemetry
+SOMNI_MONGO_RECORDS_COLLECTION=somni_records
+SOMNI_MONGO_SLEEP_REPORTS_COLLECTION=somni_sleep_reports
+SOMNI_MONGO_EVENTS_COLLECTION=somni_events
 
 # 检索参数
 SIM_THRESHOLD=0.7

app/core/config.py

@@ -53,6 +53,11 @@ class Settings(BaseSettings):
     somni_mongo_materials_collection: str = "somni_audio_materials"
     somni_mongo_tag_dictionary_collection: str = "somni_audio_tag_dictionary"
     somni_mongo_answers_collection: str = "somni_quiz_answers"
+    somni_mongo_devices_collection: str = "somni_devices"
+    somni_mongo_telemetry_collection: str = "somni_telemetry"
+    somni_mongo_records_collection: str = "somni_records"
+    somni_mongo_sleep_reports_collection: str = "somni_sleep_reports"
+    somni_mongo_events_collection: str = "somni_events"
 
     sim_threshold: float = 0.7  # 内容形态向量模糊命中阈值（规范 §五-2）
     # GetAudio query_text 与内容形态标签（含二级）向量相似度下限

app/main.py

@@ -180,7 +180,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         logger.warning("未配置 SOMNI_MONGO_URI，量产问卷与音频查询将不可用")
 
     _app_state.somni_quiz_service = SomniQuizService(somni_mongo, settings)
-    _app_state.somni_report_service = SomniReportService()
+    _app_state.somni_report_service = SomniReportService(somni_mongo, settings)
     somni_es_client = create_es_client(
         settings,
         node=settings.effective_somni_es_node,

app/server/somni/report/calc.py

@@ -0,0 +1,114 @@
+"""量产报告纯计算：本地日窗口、卧床/阶段分钟、均值 floor。"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable
+from datetime import UTC, date, datetime, timedelta
+from typing import Any
+from zoneinfo import ZoneInfo
+
+REPORT_TZ = ZoneInfo("Asia/Shanghai")
+
+
+def parse_record_date(record_date: str) -> date:
+    return date.fromisoformat(record_date)
+
+
+def local_day_utc_range(record_date: str) -> tuple[datetime, datetime]:
+    """record_date 本地日 [00:00, 次日 00:00) 转 UTC 感知时间。"""
+    day = parse_record_date(record_date)
+    start_local = datetime(day.year, day.month, day.day, tzinfo=REPORT_TZ)
+    end_local = start_local + timedelta(days=1)
+    return start_local.astimezone(UTC), end_local.astimezone(UTC)
+
+
+def ensure_aware(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=UTC)
+    return value
+
+
+def to_report_tz(value: datetime) -> datetime:
+    return ensure_aware(value).astimezone(REPORT_TZ)
+
+
+def minutes_between(start: datetime | None, end: datetime | None) -> int:
+    if start is None or end is None:
+        return 0
+    delta = to_report_tz(end) - to_report_tz(start)
+    return max(0, math.floor(delta.total_seconds() / 60))
+
+
+def stage_minutes(bed_minutes: int, ratio: int | float | None) -> int:
+    if bed_minutes <= 0 or ratio is None:
+        return 0
+    return math.floor(bed_minutes * float(ratio) / 100)
+
+
+def as_int(value: Any, default: int = 0) -> int:
+    if value is None:
+        return default
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def as_float(value: Any) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def floor_avg(values: Iterable[float]) -> int:
+    nums = list(values)
+    if not nums:
+        return 0
+    return math.floor(sum(nums) / len(nums))
+
+
+def floor_metric_stats(values: Iterable[float]) -> dict[str, int]:
+    nums = list(values)
+    if not nums:
+        return {"value": 0, "min": 0, "max": 0}
+    return {
+        "value": math.floor(sum(nums) / len(nums)),
+        "min": math.floor(min(nums)),
+        "max": math.floor(max(nums)),
+    }
+
+
+def format_hhmm(value: datetime | None) -> str:
+    if value is None:
+        return ""
+    return to_report_tz(value).strftime("%H:%M")
+
+
+def format_collected_at(value: datetime | None) -> str:
+    if value is None:
+        return ""
+    return to_report_tz(value).isoformat()
+
+
+def sleep_stage_parts(raw: dict[str, Any]) -> dict[str, Any]:
+    bed = minutes_between(raw.get("bed_time"), raw.get("wake_up_time"))
+    awake_ratio = as_int(raw.get("awake_ratio"))
+    rem_ratio = as_int(raw.get("rem_ratio"))
+    light_ratio = as_int(raw.get("light_sleep_ratio"))
+    deep_ratio = as_int(raw.get("deep_sleep_ratio"))
+    awake = stage_minutes(bed, awake_ratio)
+    rem = stage_minutes(bed, rem_ratio)
+    light = stage_minutes(bed, light_ratio)
+    deep = stage_minutes(bed, deep_ratio)
+    return {
+        "bed_minutes": bed,
+        "awake": {"minutes": awake, "percent": awake_ratio},
+        "rem_sleep": {"minutes": rem, "percent": rem_ratio},
+        "light_sleep": {"minutes": light, "percent": light_ratio},
+        "deep_sleep": {"minutes": deep, "percent": deep_ratio},
+        "total_minutes": deep + light + rem,
+    }

app/server/somni/report/rpc.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from app.core.exceptions import ServiceNotReadyError
 from app.server.errors import abort_from_app_error, abort_invalid, run_rpc_call
@@ -11,46 +11,189 @@
 if TYPE_CHECKING:
     from app.server.somni.report.service import ReportService
 
-_RPC_RES = {
-    "get_summary": uburnode_somni_pb2.GetSummaryRes,
-    "get_events": uburnode_somni_pb2.GetEventsRes,
-    "get_environment": uburnode_somni_pb2.GetEnvironmentRes,
-    "get_structure": uburnode_somni_pb2.GetStructureRes,
-    "get_sleep_quality": uburnode_somni_pb2.GetSleepQualityRes,
-}
-
 
 class ReportRpc(uburnode_somni_pb2_grpc.ReportServiceServicer):
     def __init__(self, service: ReportService | None) -> None:
         self._service = service
 
     async def GetSummary(self, request, context):
-        return await self._call(request, context, "get_summary")
+        return await self._call(request, context, "get_summary", _to_summary_res)
 
     async def GetEvents(self, request, context):
-        return await self._call(request, context, "get_events")
+        return await self._call(request, context, "get_events", _to_events_res)
 
     async def GetEnvironment(self, request, context):
-        return await self._call(request, context, "get_environment")
+        return await self._call(request, context, "get_environment", _to_environment_res)
 
     async def GetStructure(self, request, context):
-        return await self._call(request, context, "get_structure")
+        return await self._call(request, context, "get_structure", _to_structure_res)
 
     async def GetSleepQuality(self, request, context):
-        return await self._call(request, context, "get_sleep_quality")
+        return await self._call(
+            request, context, "get_sleep_quality", _to_sleep_quality_res
+        )
 
-    async def _call(self, request, context, method_name: str):
+    async def _call(self, request, context, method_name: str, to_res):
         if not request.uid.strip() or not request.record_date.strip():
             await abort_invalid(context, "uid 与 record_date 均不能为空")
         service = await self._require(context)
 
         async def _do():
-            await getattr(service, method_name)(request.uid, request.record_date)
-            return _RPC_RES[method_name]()
+            payload = await getattr(service, method_name)(
+                request.uid, request.record_date
+            )
+            return to_res(payload)
 
         return await run_rpc_call(context, _do)
 
     async def _require(self, context) -> ReportService:
         if self._service is None:
             await abort_from_app_error(context, ServiceNotReadyError())
         return self._service  # type: ignore[return-value]
+
+
+def _to_summary_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetSummaryRes:
+    item = payload.get("sleep_summary") or {}
+    return uburnode_somni_pb2.GetSummaryRes(
+        sleep_summary=uburnode_somni_pb2.SleepSummary(
+            body_battery=int(item.get("body_battery") or 0),
+            body_battery_status=str(item.get("body_battery_status") or ""),
+            total_minutes=int(item.get("total_minutes") or 0),
+            deep_sleep_minutes=int(item.get("deep_sleep_minutes") or 0),
+            avg_heart_rate=int(item.get("avg_heart_rate") or 0),
+            avg_respiratory_rate=int(item.get("avg_respiratory_rate") or 0),
+        )
+    )
+
+
+def _to_environment_res(
+    payload: dict[str, Any],
+) -> uburnode_somni_pb2.GetEnvironmentRes:
+    summary = payload.get("environment_summary") or {}
+    return uburnode_somni_pb2.GetEnvironmentRes(
+        environment_summary=uburnode_somni_pb2.EnvironmentSummary(
+            temperature=_env_metric(summary.get("temperature")),
+            humidity=_env_metric(summary.get("humidity")),
+            illuminance=_env_metric(summary.get("illuminance")),
+            noise=_env_metric(summary.get("noise")),
+        )
+    )
+
+
+def _env_metric(item: Any) -> uburnode_somni_pb2.EnvMetric:
+    data = item if isinstance(item, dict) else {}
+    return uburnode_somni_pb2.EnvMetric(
+        value=int(data.get("value") or 0),
+        min=int(data.get("min") or 0),
+        max=int(data.get("max") or 0),
+    )
+
+
+def _to_structure_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetStructureRes:
+    structure = payload.get("sleep_structure") or {}
+    return uburnode_somni_pb2.GetStructureRes(
+        sleep_structure=uburnode_somni_pb2.SleepStructure(
+            awake=_stage_part(structure.get("awake")),
+            rem_sleep=_stage_part(structure.get("rem_sleep")),
+            light_sleep=_stage_part(structure.get("light_sleep")),
+            deep_sleep=_stage_part(structure.get("deep_sleep")),
+        )
+    )
+
+
+def _stage_part(item: Any) -> uburnode_somni_pb2.SleepStagePart:
+    data = item if isinstance(item, dict) else {}
+    return uburnode_somni_pb2.SleepStagePart(
+        minutes=int(data.get("minutes") or 0),
+        percent=int(data.get("percent") or 0),
+    )
+
+
+def _to_sleep_quality_res(
+    payload: dict[str, Any],
+) -> uburnode_somni_pb2.GetSleepQualityRes:
+    item = payload.get("sleep_quality") or {}
+    return uburnode_somni_pb2.GetSleepQualityRes(
+        sleep_quality=uburnode_somni_pb2.SleepQuality(
+            time_in_bed_minutes=int(item.get("time_in_bed_minutes") or 0),
+            sleep_onset_latency_minutes=int(
+                item.get("sleep_onset_latency_minutes") or 0
+            ),
+            sleep_efficiency=int(item.get("sleep_efficiency") or 0),
+            bedtime=str(item.get("bedtime") or ""),
+            wake_up_time=str(item.get("wake_up_time") or ""),
+            awake_after_onset_minutes=int(item.get("awake_after_onset_minutes") or 0),
+        )
+    )
+
+
+def _to_events_res(payload: dict[str, Any]) -> uburnode_somni_pb2.GetEventsRes:
+    res = uburnode_somni_pb2.GetEventsRes(
+        record_date=str(payload.get("record_date") or ""),
+        event_count=int(payload.get("event_count") or 0),
+        abnormal_count=int(payload.get("abnormal_count") or 0),
+        intervention_count=int(payload.get("intervention_count") or 0),
+    )
+    for item in payload.get("sleep_events") or []:
+        res.sleep_events.append(_sleep_event_item(item))
+    for item in payload.get("idf_data") or []:
+        res.idf_data.append(
+            uburnode_somni_pb2.IdfStage(
+                stage=str(item.get("stage") or ""),
+                start=str(item.get("start") or ""),
+                end=str(item.get("end") or ""),
+            )
+        )
+    for item in payload.get("physio_data") or []:
+        metrics = item.get("metrics") or {}
+        res.physio_data.append(
+            uburnode_somni_pb2.PhysioDataPoint(
+                collected_at=str(item.get("collected_at") or ""),
+                metrics=uburnode_somni_pb2.PhysioMetrics(
+                    heart_rate=int(metrics.get("heart_rate") or 0),
+                    respiration_rate=int(metrics.get("respiration_rate") or 0),
+                ),
+            )
+        )
+    for item in payload.get("env_data") or []:
+        res.env_data.append(
+            uburnode_somni_pb2.EnvDataPoint(
+                collected_at=str(item.get("collected_at") or ""),
+                temperature=int(item.get("temperature") or 0),
+                humidity=int(item.get("humidity") or 0),
+                illuminance=int(item.get("illuminance") or 0),
+                noise=int(item.get("noise") or 0),
+            )
+        )
+    return res
+
+
+def _sleep_event_item(item: dict[str, Any]) -> uburnode_somni_pb2.SleepEventItem:
+    event = uburnode_somni_pb2.SleepEventItem(
+        event_time=str(item.get("event_time") or ""),
+        type=str(item.get("type") or ""),
+        code=str(item.get("code") or ""),
+    )
+    for detail in item.get("events") or []:
+        event.events.append(
+            uburnode_somni_pb2.SleepEventDetail(
+                event_type=str(detail.get("event_type") or ""),
+                duration=str(detail.get("duration") or ""),
+                trigger_cause=str(detail.get("trigger_cause") or ""),
+                action_taken=str(detail.get("action_taken") or ""),
+                result_summary=str(detail.get("result_summary") or ""),
+            )
+        )
+    intervention = item.get("intervention") or {}
+    event.intervention.CopyFrom(
+        uburnode_somni_pb2.Intervention(
+            type=str(intervention.get("type") or ""),
+            event_time=str(intervention.get("event_time") or ""),
+            event_type=str(intervention.get("event_type") or ""),
+            duration=str(intervention.get("duration") or ""),
+            trigger_cause=str(intervention.get("trigger_cause") or ""),
+            action_taken=str(intervention.get("action_taken") or ""),
+            result_summary=str(intervention.get("result_summary") or ""),
+        )
+    )
+    return event