Grigo/Home_assistant
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

smart tdee

Browse Source
This commit is contained in:
Grigo
2026-06-16 04:38:23 +00:00
parent f2e98942ff
commit a3f01cd850
56 changed files with 2519 additions and 591 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/chat.py
+102 -8
View File
@@ -1,6 +1,7 @@
import asyncio
import json
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
@@ -11,6 +12,7 @@ from app.api.schemas import (
SessionDetailOut,
SessionOut,
)
from app.auth.deps import get_current_user
from app.chat.generation import (
GenerationBusyError,
get_active_handle,
@@ -19,12 +21,18 @@ from app.chat.generation import (
subscribe_generation,
)
from app.chat.service import ChatService
from app.auth.deps import get_current_user
from app.config import get_settings
from app.db.base import get_db
from app.db.models import User
from app.vision import VisionService, format_user_messages, vision_debug_payloads
from app.vision.analyze import VisionUnavailableError
from app.vision.preprocess import prepare_image
from app.vision.storage import format_upload_images_markdown, save_upload
router = APIRouter()
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
@router.post("/sessions", response_model=SessionOut)
def create_session(payload: SessionCreate, db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> SessionOut:
@@ -108,11 +116,95 @@ def delete_session(session_id: int, db: Session = Depends(get_db), user: User =
return {"ok": True}
def _collect_form_uploads(form) -> list:
uploads: list = []
seen_ids: set[int] = set()
def _append(item) -> None:
if item is None or not hasattr(item, "read"):
return
item_id = id(item)
if item_id in seen_ids:
return
seen_ids.add(item_id)
uploads.append(item)
if hasattr(form, "getlist"):
for item in form.getlist("images"):
_append(item)
single = form.get("image")
_append(single)
return uploads
async def _analyze_upload(raw: bytes, *, caption: str, user_id: int):
prepared = prepare_image(raw)
filename = save_upload(prepared, user_id=user_id)
result = await VisionService().analyze_prepared(prepared, user_hint=caption)
return result, filename
async def _parse_message_request(
request: Request,
*,
user_id: int,
) -> tuple[str, dict | None]:
content_type = (request.headers.get("content-type") or "").lower()
if "multipart/form-data" not in content_type:
try:
body = await request.json()
except json.JSONDecodeError as exc:
raise HTTPException(status_code=400, detail="Invalid JSON body") from exc
payload = MessageCreate.model_validate(body)
return payload.content, None
form = await request.form()
caption = str(form.get("content") or "").strip()
uploads = _collect_form_uploads(form)
if not uploads:
raise HTTPException(status_code=400, detail="Field 'images' or 'image' is required for multipart upload")
max_images = max(1, int(get_settings().vision_max_images))
if len(uploads) > max_images:
raise HTTPException(
status_code=400,
detail=f"Too many images (max {max_images})",
)
raw_images: list[bytes] = []
for upload in uploads:
raw = await upload.read()
if not raw:
raise HTTPException(status_code=400, detail="Empty image file")
mime = getattr(upload, "content_type", None) or "application/octet-stream"
if mime not in ALLOWED_IMAGE_TYPES:
raise HTTPException(status_code=400, detail=f"Unsupported image type: {mime}")
raw_images.append(raw)
try:
analyzed = await asyncio.gather(
*(_analyze_upload(raw, caption=caption, user_id=user_id) for raw in raw_images)
)
except VisionUnavailableError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
results = [item[0] for item in analyzed]
filenames = [item[1] for item in analyzed]
debug = vision_debug_payloads(results)
vision_text = format_user_messages(caption, results)
images_md = format_upload_images_markdown(user_id, filenames)
user_text = f"{images_md}\n\n{vision_text}" if images_md else vision_text
if not user_text.strip():
raise HTTPException(status_code=400, detail="Could not build message from image")
return user_text, debug
@router.post("/sessions/{session_id}/messages")
async def send_message(
session_id: int,
payload: MessageCreate,
db: Session = Depends(get_db), user: User = Depends(get_current_user),
request: Request,
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
) -> StreamingResponse:
service = ChatService(db, user.id)
if not service.get_session(session_id):
@@ -121,16 +213,19 @@ async def send_message(
if is_generation_active(session_id):
raise HTTPException(status_code=409, detail="Generation already in progress")
# Сохраняем user до стрима: иначе при обрыве SSE сообщение не попадает в БД.
service.save_user_message(session_id, payload.content)
user_text, vision_debug = await _parse_message_request(request, user_id=user.id)
service.save_user_message(session_id, user_text)
try:
handle = await start_generation(session_id, user.id, payload.content)
handle = await start_generation(session_id, user.id, user_text)
except GenerationBusyError:
raise HTTPException(status_code=409, detail="Generation already in progress") from None
async def event_stream():
try:
if vision_debug:
yield ChatService._sse("vision", vision_debug)
async for chunk in subscribe_generation(handle):
yield chunk
except asyncio.CancelledError:
@@ -155,4 +250,3 @@ def context_preview(
) -> dict:
service = ChatService(db, user.id)
return service.context_preview(session_id, query=query)
backend/app/api/routes/fitness.py
+3 -4
View File
@@ -21,12 +21,9 @@ class ProfileUpdate(BaseModel):
age: int | None = None
height_cm: float | None = None
weight_kg: float | None = None
activity_level: str | None = None
goal: str | None = None
target_weight_kg: float | None = None
weekly_workouts: int | None = None
baseline_steps: int | None = None
baseline_workout_kcal: float | None = None
neat_base_kcal: float | None = Field(default=None, ge=200, le=300)
class MealCreate(BaseModel):
@@ -254,6 +251,8 @@ async def create_workout(
active_calories=structured.get("active_calories"),
total_calories=structured.get("total_calories"),
steps=structured.get("steps"),
activity_type=structured.get("activity_type"),
met=structured.get("met"),
day=day,
days_ago=payload.days_ago,
logged_at=payload.logged_at,
backend/app/api/routes/homelab.py
+4 -2
View File
@@ -48,7 +48,9 @@ def homelab_status() -> dict:
@router.get("/weather")
def weather_dashboard(
hours_ahead: int = 12,
days_ahead: int = 7,
_: User = Depends(get_current_user),
) -> dict:
hours = max(1, min(int(hours_ahead), 48))
return build_weather_dashboard(hours_ahead=hours)
hours = max(1, min(int(hours_ahead), 168))
days = max(1, min(int(days_ahead), 16))
return build_weather_dashboard(hours_ahead=hours, days_ahead=days)
backend/app/api/routes/media.py
+22 -1
View File
@@ -1,9 +1,11 @@
from pathlib import Path
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import FileResponse
from app.auth.deps import get_current_user
from app.config import get_settings
from app.db.models import User
router = APIRouter(prefix="/media", tags=["media"])
@@ -19,3 +21,22 @@ def get_generated_image(filename: str) -> FileResponse:
raise HTTPException(status_code=404, detail="File not found")
return FileResponse(path, media_type="image/png")
@router.get("/uploads/{user_id}/{filename}")
def get_upload_image(
user_id: int,
filename: str,
user: User = Depends(get_current_user),
) -> FileResponse:
if user.id != user_id:
raise HTTPException(status_code=403, detail="Forbidden")
if ".." in filename or "/" in filename or "\\" in filename:
raise HTTPException(status_code=400, detail="Invalid filename")
settings = get_settings()
path = Path(settings.uploads_dir) / str(user_id) / filename
if not path.is_file():
raise HTTPException(status_code=404, detail="File not found")
return FileResponse(path, media_type="image/jpeg")
backend/app/api/routes/settings.py
+1
View File
@@ -15,6 +15,7 @@ router = APIRouter()
class SettingsPatch(BaseModel):
openrouter_model: str | None = None
memory_extract_model: str | None = None
openrouter_vision_model: str | None = None
openrouter_reasoning_effort: str | None = None
rag_enabled: bool | None = None
rag_top_k: int | None = Field(default=None, ge=1, le=50)
backend/app/auth/deps.py
+4 -1
View File
@@ -14,7 +14,10 @@ def _extract_token(request: Request) -> str | None:
if token:
return token
header = request.headers.get("X-API-Token", "").strip()
return header or None
if header:
return header
query = request.query_params.get("token", "").strip()
return query or None
def get_current_user(
backend/app/character/card.py
+4
View File
@@ -14,6 +14,10 @@ TOOLS_INSTRUCTIONS = """
- create_work_item — при «заведи баг/фичу»; передай полный текст и project_slug.
- Фитнес: get_fitness_summary (date/days_ago), get_fitness_history, set_fitness_profile, log_meal, log_water, log_weight (neck_cm/waist_cm/hip_cm → Navy), log_workout,
- «Что ел вчера» → get_fitness_summary days_ago=1. «За неделю» → get_fitness_history.
- Скриншоты и фото: vision-модель уже разобрала каждую картинку ДО твоего ответа. В сообщении один или несколько блоков [Скриншот] / [Скриншот N/M] — это содержимое изображений; отвечай так, будто ты их видишь.
- НЕ говори, что у тебя нет глаз / ты не видишь картинку / нужен Gemini, OpenRouter или curl — распознавание уже выполнено.
- fitness_workout / fitness_steps + fitness_hints: log_workout, log_steps и т.д.; при confidence=low уточни детали.
- document_type=other: опиши и прокомментируй по блоку [Скриншот], без советов про настройку vision API.
calc_fitness_targets, calc_body_composition (расчёт Navy/WHR/LBM/FFMI без записи), lookup_food, lookup_exercise, set_fitness_reminder.
- Память: remember_fact, recall_memories, forget_memory, update_profile, update_session_summary.
- «Запомни» → remember_fact. «Кто я» / «сколько мне лет» → профиль и факты из блока [Память], не выдумывай.
backend/app/chat/service.py
+16 -2
View File
@@ -20,7 +20,7 @@ from app.chat.notices import (
)
from app.fitness.context import format_fitness_context, get_fitness_snapshot
from app.homelab.context import format_datetime_context
from app.homelab.openmeteo import format_weather_snapshot
from app.homelab.openmeteo import OpenMeteoClient, format_weather_snapshot
from app.memory.context import (
format_identity_hint,
format_memory_context,
@@ -34,6 +34,7 @@ from app.db.models import ChatSession, Message
from app.llm.client import LLMClient
from app.pomodoro.service import PomodoroService
from app.tools.registry import TOOL_DEFINITIONS, execute_tool
from app.vision.analyze import format_vision_turn_hint
MAX_TOOL_ROUNDS = 5
MAX_HISTORY_MESSAGES = 40
@@ -45,6 +46,11 @@ _DOMAIN_KEYWORDS: dict[str, tuple[str, ...]] = {
"shopping": ("покуп", "магазин", "список", "shopping", "корзин"),
"reminders": ("напомин", "календар", "событи", "дедлайн", "встреч", "план"),
"projects": ("taiga", "gitea", "задач", "проект", "git", "issue", "коммит", "ветк"),
"weather": (
"погод", "дожд", "снег", "ветер", "температур", "градус", "мороз", "жар",
"на улице", "одеть", "зонт", "прогноз", "завтра", "послезавтра", "выходн",
"weather", "rain", "forecast", "umbrella", "outside",
),
}
logger = logging.getLogger(__name__)
@@ -186,7 +192,12 @@ class ChatService:
self._optional_domain("fitness", user_query, lambda: fitness_snapshot, format_fitness_context),
self._optional_domain("shopping", user_query, lambda: shopping_snapshot, format_shopping_context),
self._optional_domain("reminders", user_query, lambda: reminders_snapshot, format_reminders_context),
format_weather_snapshot(),
self._optional_domain(
"weather",
user_query,
lambda: OpenMeteoClient().fetch_forecast(hours_ahead=6, days_ahead=7),
lambda snap: format_weather_snapshot(snap, include_daily=True),
),
format_pomodoro_context(status),
self._optional_domain("projects", user_query, lambda: projects_snapshot, format_projects_context),
]
@@ -201,6 +212,9 @@ class ChatService:
identity_hint = format_identity_hint(memory_snapshot, last_user)
if identity_hint:
system_prompt += f"\n\n{identity_hint}"
vision_hint = format_vision_turn_hint(last_user)
if vision_hint:
system_prompt += f"\n\n{vision_hint}"
if len(all_chat) > MAX_HISTORY_MESSAGES:
system_prompt += (
f"\n\n[История чата: в контексте последние {MAX_HISTORY_MESSAGES} "
backend/app/config.py
+23
View File
@@ -1,8 +1,19 @@
from functools import lru_cache
from pathlib import Path
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
DEPRECATED_VISION_MODELS: dict[str, str] = {
"google/gemini-2.0-flash-lite-001": "google/gemini-2.5-flash-lite",
"google/gemini-2.0-flash-lite": "google/gemini-2.5-flash-lite",
}
def resolve_vision_model(model: str) -> str:
stripped = model.strip()
return DEPRECATED_VISION_MODELS.get(stripped, stripped)
class Settings(BaseSettings):
model_config = SettingsConfigDict(
@@ -23,6 +34,17 @@ class Settings(BaseSettings):
openrouter_tools_enabled: bool = True
# DeepSeek V4 / reasoning: none | low | medium | high | xhigh. none = без thinking.
openrouter_reasoning_effort: str = "none"
openrouter_vision_model: str = "google/gemini-2.5-flash-lite"
vision_max_edge_px: int = 1280
vision_jpeg_quality: int = 85
vision_debug_enabled: bool = True
vision_max_images: int = 8
uploads_dir: str = "./data/uploads"
@field_validator("openrouter_vision_model")
@classmethod
def migrate_vision_model(cls, value: str) -> str:
return resolve_vision_model(value)
database_url: str = "sqlite:///./data/assistant.db"
cors_origins: str = "http://localhost:5173,http://localhost:8080,http://localhost:3000"
@@ -63,6 +85,7 @@ class Settings(BaseSettings):
weather_lon: float = 30.3351
weather_location_name: str = "Санкт-Петербург"
weather_cache_sec: int = 300
weather_forecast_days: int = 7
openmeteo_fallback_url: str = "https://api.open-meteo.com"
openmeteo_fallback_on_partial: bool = True
backend/app/db/migrate_fitness.py
+130 -1
View File
@@ -1,6 +1,20 @@
from sqlalchemy import inspect, text
import logging
from sqlalchemy import inspect, select, text
from sqlalchemy.orm import Session
from app.db.base import engine
from app.db.models import FitnessProfile
from app.fitness.calculators import DEFAULT_NEAT_KCAL, compute_targets, macro_targets
logger = logging.getLogger(__name__)
TDEE_V2_BACKFILL = "fitness_tdee_v2_backfill"
MACROS_GKG_BACKFILL = "fitness_macros_gkg_v1"
def _table_exists(table: str) -> bool:
return table in inspect(engine).get_table_names()
def _add_column_if_missing(table: str, column: str, ddl: str) -> None:
@@ -14,6 +28,113 @@ def _add_column_if_missing(table: str, column: str, ddl: str) -> None:
conn.execute(text(ddl))
def _ensure_schema_migrations_table() -> None:
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE IF NOT EXISTS _schema_migrations ("
"name TEXT PRIMARY KEY, "
"applied_at DATETIME DEFAULT CURRENT_TIMESTAMP)"
)
)
def _migration_applied(name: str) -> bool:
_ensure_schema_migrations_table()
with engine.begin() as conn:
row = conn.execute(
text("SELECT 1 FROM _schema_migrations WHERE name = :name"),
{"name": name},
).fetchone()
return row is not None
def _mark_migration_applied(name: str) -> None:
with engine.begin() as conn:
conn.execute(
text("INSERT INTO _schema_migrations (name) VALUES (:name)"),
{"name": name},
)
def _profile_targets(row: FitnessProfile) -> dict[str, float]:
neat = row.neat_base_kcal if row.neat_base_kcal is not None else DEFAULT_NEAT_KCAL
return compute_targets(
{
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"goal": row.goal,
"neat_base_kcal": neat,
}
)
def backfill_tdee_targets(*, force: bool = False) -> int:
"""Recalculate stored calorie/macro targets for all profiles (PAL → BMR+NEAT)."""
if not _table_exists("fitness_profiles"):
return 0
_ensure_schema_migrations_table()
if not force and _migration_applied(TDEE_V2_BACKFILL):
return 0
with engine.begin() as conn:
conn.execute(
text(
"UPDATE fitness_profiles "
"SET neat_base_kcal = :neat "
"WHERE neat_base_kcal IS NULL"
),
{"neat": DEFAULT_NEAT_KCAL},
)
updated = 0
with Session(engine) as db:
rows = db.scalars(select(FitnessProfile)).all()
for row in rows:
if row.neat_base_kcal is None:
row.neat_base_kcal = DEFAULT_NEAT_KCAL
targets = _profile_targets(row)
row.calorie_target = targets["calorie_target"]
row.protein_g = targets["protein_g"]
row.fat_g = targets["fat_g"]
row.carbs_g = targets["carbs_g"]
row.water_l = targets["water_l"]
updated += 1
db.commit()
if not force or not _migration_applied(TDEE_V2_BACKFILL):
_mark_migration_applied(TDEE_V2_BACKFILL)
logger.info("TDEE v2 backfill: recalculated %s fitness profile(s)", updated)
return updated
def backfill_macros_gkg(*, force: bool = False) -> int:
"""Recalculate stored BJU from weight (protein/fat g/kg, carbs = remainder)."""
if not _table_exists("fitness_profiles"):
return 0
_ensure_schema_migrations_table()
if not force and _migration_applied(MACROS_GKG_BACKFILL):
return 0
updated = 0
with Session(engine) as db:
rows = db.scalars(select(FitnessProfile)).all()
for row in rows:
macros = macro_targets(row.calorie_target, row.weight_kg, row.goal)
row.protein_g = macros["protein_g"]
row.fat_g = macros["fat_g"]
row.carbs_g = macros["carbs_g"]
updated += 1
db.commit()
_mark_migration_applied(MACROS_GKG_BACKFILL)
logger.info("Macros g/kg backfill: updated %s fitness profile(s)", updated)
return updated
def run_fitness_migrations() -> None:
inspector = inspect(engine)
@@ -28,6 +149,11 @@ def run_fitness_migrations() -> None:
"baseline_workout_kcal",
"ALTER TABLE fitness_profiles ADD COLUMN baseline_workout_kcal FLOAT",
)
_add_column_if_missing(
"fitness_profiles",
"neat_base_kcal",
"ALTER TABLE fitness_profiles ADD COLUMN neat_base_kcal FLOAT DEFAULT 200.0",
)
if "workout_logs" in inspector.get_table_names():
_add_column_if_missing(
@@ -92,3 +218,6 @@ def run_fitness_migrations() -> None:
"ffmi",
"ALTER TABLE body_metrics ADD COLUMN ffmi FLOAT",
)
backfill_tdee_targets()
backfill_macros_gkg()
backend/app/db/models.py
+1
View File
@@ -179,6 +179,7 @@ class FitnessProfile(Base):
activity_level: Mapped[str] = mapped_column(String(32), default="moderate")
goal: Mapped[str] = mapped_column(String(32), default="maintain")
target_weight_kg: Mapped[float | None] = mapped_column(Float, nullable=True)
neat_base_kcal: Mapped[float] = mapped_column(Float, default=200.0)
weekly_workouts: Mapped[int] = mapped_column(Integer, default=3)
baseline_steps: Mapped[int | None] = mapped_column(Integer, nullable=True)
baseline_workout_kcal: Mapped[float | None] = mapped_column(Float, nullable=True)
backend/app/fitness/activity_budget.py
+53 -128
View File
@@ -1,143 +1,68 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from typing import Any
BASELINE_STEPS_BY_LEVEL: dict[str, int] = {
"sedentary": 5000,
"light": 7000,
"moderate": 9000,
"active": 11000,
"very_active": 13000,
}
DEFAULT_MET = 5.0
WORKOUT_KCAL_PER_SESSION = 200
KCAL_PER_STEP_PER_KG = 0.0005
FALLBACK_KCAL_PER_MIN = 6
MET_BY_KEYWORD: list[tuple[str, float]] = [
("триатлон", 10.0),
("марафон", 9.8),
("бег", 9.8),
("running", 9.8),
("run", 9.0),
("плаван", 8.0),
("swim", 8.0),
("велосипед", 7.5),
("cycling", 7.5),
("вел", 7.5),
("hiit", 8.0),
("кроссфит", 8.0),
("силов", 6.0),
("strength", 6.0),
("зал", 5.5),
("gym", 5.5),
("йога", 3.0),
("yoga", 3.0),
("ходьб", 3.5),
("walk", 3.5),
("прогул", 3.5),
]
@dataclass
class ActivityBonus:
steps: int
steps_baseline: int
steps_bonus_kcal: float
workout_active_kcal: float
workout_baseline_kcal: float
workout_bonus_kcal: float
total_bonus_kcal: float
scale_factor: float
def infer_met(workout: dict[str, Any]) -> float | None:
explicit = workout.get("met")
if explicit is not None:
return float(explicit)
def to_dict(self) -> dict[str, Any]:
return asdict(self)
activity_type = str(workout.get("activity_type") or "").lower()
title = str(workout.get("title") or "").lower()
notes = str(workout.get("notes") or "").lower()
haystack = f"{activity_type} {title} {notes}"
for keyword, met in MET_BY_KEYWORD:
if keyword in haystack:
return met
return None
def baseline_steps(profile: dict[str, Any]) -> int:
override = profile.get("baseline_steps")
if override is not None:
return int(override)
level = str(profile.get("activity_level") or "moderate")
return BASELINE_STEPS_BY_LEVEL.get(level, 9000)
def baseline_workout_kcal_day(profile: dict[str, Any]) -> float:
override = profile.get("baseline_workout_kcal")
if override is not None:
return float(override)
weekly = int(profile.get("weekly_workouts") or 3)
return round(weekly * WORKOUT_KCAL_PER_SESSION / 7, 1)
def estimate_workout_active_kcal(workout: dict[str, Any]) -> float:
def estimate_workout_active_kcal(workout: dict[str, Any], *, weight_kg: float) -> float:
active = workout.get("active_calories")
if active is not None:
return float(active)
return round(float(active), 1)
duration = workout.get("duration_min")
if duration:
return float(duration) * FALLBACK_KCAL_PER_MIN
return 0.0
if not duration:
return 0.0
met = infer_met(workout)
if met is None:
return 0.0
hours = float(duration) / 60.0
return round(met * weight_kg * hours, 1)
def steps_bonus_kcal(*, steps: int, baseline_steps: int, weight_kg: float) -> float:
extra_steps = max(0, steps - baseline_steps)
return round(extra_steps * weight_kg * KCAL_PER_STEP_PER_KG, 1)
def compute_activity_bonus(
profile: dict[str, Any],
*,
steps_total: int,
workouts: list[dict[str, Any]],
) -> ActivityBonus:
weight_kg = float(profile.get("weight_kg") or 70)
steps_base = baseline_steps(profile)
workout_base = baseline_workout_kcal_day(profile)
s_bonus = steps_bonus_kcal(steps=steps_total, baseline_steps=steps_base, weight_kg=weight_kg)
workout_active = round(sum(estimate_workout_active_kcal(w) for w in workouts), 1)
w_bonus = max(0.0, round(workout_active - workout_base, 1))
total_bonus = round(s_bonus + w_bonus, 1)
base_cal = float(profile.get("calorie_target") or 2000)
scale_factor = 1.0 if base_cal <= 0 else round((base_cal + total_bonus) / base_cal, 4)
return ActivityBonus(
steps=steps_total,
steps_baseline=steps_base,
steps_bonus_kcal=s_bonus,
workout_active_kcal=workout_active,
workout_baseline_kcal=workout_base,
workout_bonus_kcal=w_bonus,
total_bonus_kcal=total_bonus,
scale_factor=scale_factor,
)
def _targets_dict(
*,
calories: float,
protein_g: float,
fat_g: float,
carbs_g: float,
water_ml: float,
) -> dict[str, float]:
return {
"calories": round(calories),
"protein_g": round(protein_g),
"fat_g": round(fat_g),
"carbs_g": round(carbs_g),
"water_ml": round(water_ml),
}
def build_base_targets(profile: dict[str, Any]) -> dict[str, float]:
water_l = float(profile.get("water_l") or 2.5)
return _targets_dict(
calories=float(profile.get("calorie_target") or 2000),
protein_g=float(profile.get("protein_g") or 140),
fat_g=float(profile.get("fat_g") or 65),
carbs_g=float(profile.get("carbs_g") or 200),
water_ml=water_l * 1000,
)
def scale_targets(
base_targets: dict[str, float],
bonus_kcal: float,
) -> tuple[dict[str, float], dict[str, float]]:
"""Return (effective_targets, targets_base). Water is not scaled."""
targets_base = dict(base_targets)
base_cal = float(base_targets["calories"])
if bonus_kcal <= 0 or base_cal <= 0:
return dict(base_targets), targets_base
scale = (base_cal + bonus_kcal) / base_cal
effective = _targets_dict(
calories=base_cal + bonus_kcal,
protein_g=float(base_targets["protein_g"]) * scale,
fat_g=float(base_targets["fat_g"]) * scale,
carbs_g=float(base_targets["carbs_g"]) * scale,
water_ml=float(base_targets["water_ml"]),
)
return effective, targets_base
def workouts_kcal_total(workouts: list[dict[str, Any]], *, weight_kg: float) -> float:
if not workouts:
return 0.0
return round(sum(estimate_workout_active_kcal(w, weight_kg=weight_kg) for w in workouts), 1)
backend/app/fitness/calculators.py
+104 -30
View File
@@ -1,12 +1,12 @@
from typing import Any
ACTIVITY_MULTIPLIERS = {
"sedentary": 1.2,
"light": 1.375,
"moderate": 1.55,
"active": 1.725,
"very_active": 1.9,
}
from app.fitness.activity_budget import workouts_kcal_total
DEFAULT_NEAT_KCAL = 200.0
NEAT_KCAL_MIN = 200.0
NEAT_KCAL_MAX = 300.0
KCAL_PER_STEP_REF = 0.04 / 86 # ~0.04 kcal/step at 86 kg
WATER_ML_PER_KG = 33 # middle of 30–35 ml/kg range
GOAL_CALORIE_ADJUST = {
"lose": -500,
@@ -14,6 +14,13 @@ GOAL_CALORIE_ADJUST = {
"gain": 300,
}
PROTEIN_G_PER_KG = {
"lose": 2.2,
"maintain": 1.8,
"gain": 1.8,
}
FAT_G_PER_KG = 1.0
def bmr_mifflin(*, sex: str, weight_kg: float, height_cm: float, age: int) -> float:
base = 10 * weight_kg + 6.25 * height_cm - 5 * age
@@ -22,17 +29,17 @@ def bmr_mifflin(*, sex: str, weight_kg: float, height_cm: float, age: int) -> fl
return base - 161
def tdee(
*,
sex: str,
weight_kg: float,
height_cm: float,
age: int,
activity_level: str = "moderate",
) -> float:
bmr = bmr_mifflin(sex=sex, weight_kg=weight_kg, height_cm=height_cm, age=age)
mult = ACTIVITY_MULTIPLIERS.get(activity_level, 1.55)
return bmr * mult
def neat_base_kcal(profile: dict[str, Any]) -> float:
raw = profile.get("neat_base_kcal")
if raw is not None:
return max(NEAT_KCAL_MIN, min(NEAT_KCAL_MAX, float(raw)))
return DEFAULT_NEAT_KCAL
def steps_kcal(*, steps: int, weight_kg: float) -> float:
if steps <= 0:
return 0.0
return round(steps * weight_kg * KCAL_PER_STEP_REF, 1)
def bmi(weight_kg: float, height_cm: float) -> float:
@@ -43,7 +50,7 @@ def bmi(weight_kg: float, height_cm: float) -> float:
def water_target_l(weight_kg: float) -> float:
return round(weight_kg * 0.033, 1)
return round(weight_kg * WATER_ML_PER_KG / 1000, 1)
def macro_targets(
@@ -51,8 +58,8 @@ def macro_targets(
weight_kg: float,
goal: str = "maintain",
) -> dict[str, float]:
protein_g = round(weight_kg * (2.0 if goal == "gain" else 1.8), 0)
fat_g = round((calorie_target * 0.25) / 9, 0)
protein_g = round(weight_kg * PROTEIN_G_PER_KG.get(goal, 1.8), 0)
fat_g = round(weight_kg * FAT_G_PER_KG, 0)
protein_cal = protein_g * 4
fat_cal = fat_g * 9
carbs_g = max(0, round((calorie_target - protein_cal - fat_cal) / 4, 0))
@@ -67,28 +74,95 @@ def one_rep_max(weight_kg: float, reps: int) -> float:
return round(weight_kg * (1 + reps / 30), 1)
def compute_targets(profile: dict[str, Any]) -> dict[str, Any]:
def _profile_fields(profile: dict[str, Any]) -> tuple[float, float, int, str, str]:
weight = float(profile.get("weight_kg") or 70)
height = float(profile.get("height_cm") or 170)
age = int(profile.get("age") or 30)
sex = str(profile.get("sex") or "male")
activity = str(profile.get("activity_level") or "moderate")
goal = str(profile.get("goal") or "maintain")
return weight, height, age, sex, goal
tdee_val = tdee(
sex=sex, weight_kg=weight, height_cm=height, age=age, activity_level=activity
)
calorie_target = round(tdee_val + GOAL_CALORIE_ADJUST.get(goal, 0), 0)
def compute_tdee(
profile: dict[str, Any],
*,
steps_total: int = 0,
workouts: list[dict[str, Any]] | None = None,
) -> dict[str, float]:
weight, height, age, sex, _ = _profile_fields(profile)
bmr = bmr_mifflin(sex=sex, weight_kg=weight, height_cm=height, age=age)
neat = neat_base_kcal(profile)
s_kcal = steps_kcal(steps=steps_total, weight_kg=weight)
w_kcal = workouts_kcal_total(workouts or [], weight_kg=weight)
tdee_val = bmr + neat + s_kcal + w_kcal
return {
"bmr": round(bmr, 0),
"neat_kcal": round(neat, 0),
"steps_kcal": s_kcal,
"workout_kcal": w_kcal,
"tdee": round(tdee_val, 0),
}
def compute_daily_targets(
profile: dict[str, Any],
*,
steps_total: int = 0,
workouts: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
weight, height, age, sex, goal = _profile_fields(profile)
breakdown = compute_tdee(profile, steps_total=steps_total, workouts=workouts)
calorie_target = round(breakdown["tdee"] + GOAL_CALORIE_ADJUST.get(goal, 0), 0)
macros = macro_targets(calorie_target, weight, goal)
water = water_target_l(weight)
return {
"bmr": round(bmr_mifflin(sex=sex, weight_kg=weight, height_cm=height, age=age), 0),
"tdee": round(tdee_val, 0),
"bmi": round(bmi(weight, height), 1),
**breakdown,
"calorie_target": calorie_target,
"protein_g": macros["protein_g"],
"fat_g": macros["fat_g"],
"carbs_g": macros["carbs_g"],
"water_l": water,
"bmi": round(bmi(weight, height), 1),
"steps": steps_total,
}
def targets_to_api(daily: dict[str, Any]) -> dict[str, float]:
return {
"calories": daily["calorie_target"],
"protein_g": daily["protein_g"],
"fat_g": daily["fat_g"],
"carbs_g": daily["carbs_g"],
"water_ml": round(daily["water_l"] * 1000),
}
def tdee_breakdown_to_api(daily: dict[str, Any]) -> dict[str, Any]:
return {
"bmr": daily["bmr"],
"neat_kcal": daily["neat_kcal"],
"steps_kcal": daily["steps_kcal"],
"workout_kcal": daily["workout_kcal"],
"tdee": daily["tdee"],
"calorie_target": daily["calorie_target"],
"steps": daily.get("steps", 0),
}
def compute_targets(profile: dict[str, Any]) -> dict[str, Any]:
"""Rest-day targets (BMR + NEAT, no steps/workouts) for profile storage."""
daily = compute_daily_targets(profile, steps_total=0, workouts=[])
return {
"bmr": daily["bmr"],
"tdee": daily["tdee"],
"bmi": daily["bmi"],
"neat_kcal": daily["neat_kcal"],
"steps_kcal": 0,
"workout_kcal": 0,
"calorie_target": daily["calorie_target"],
"protein_g": daily["protein_g"],
"fat_g": daily["fat_g"],
"carbs_g": daily["carbs_g"],
"water_l": daily["water_l"],
}
backend/app/fitness/context.py
+22 -10
View File
@@ -16,11 +16,16 @@ def format_fitness_context(snapshot: dict[str, Any]) -> str:
if not profile:
lines.append("Профиль не настроен. set_fitness_profile для целей ккал/БЖУ/воды.")
else:
computed = profile.get("computed") or {}
lines.append(
f"Цели (база): {profile.get('calorie_target')} ккал, "
f"Цели (база, без шагов/тренировок): {profile.get('calorie_target')} ккал, "
f"Б {profile.get('protein_g')} / Ж {profile.get('fat_g')} / У {profile.get('carbs_g')} г, "
f"вода {profile.get('water_l')} л"
)
lines.append(
f"BMR {computed.get('bmr', '?')} + NEAT {computed.get('neat_kcal', 200)} = "
f"TDEE база {computed.get('tdee', '?')} ккал"
)
if profile.get("goal"):
lines.append(
f"Цель: {profile.get('goal')}, вес {profile.get('weight_kg')} кг, "
@@ -30,19 +35,23 @@ def format_fitness_context(snapshot: dict[str, Any]) -> str:
today = snapshot.get("today") or {}
totals = today.get("totals") or {}
targets = today.get("targets") or {}
targets_base = today.get("targets_base") or {}
activity = today.get("activity") or {}
breakdown = today.get("tdee_breakdown") or {}
steps_total = today.get("steps_total") or 0
water_l = totals.get("water_ml", 0) / 1000
water_target = targets.get("water_ml", 2500) / 1000
if profile and (activity.get("total_bonus_kcal") or steps_total):
if breakdown:
lines.append(
f"Активность: шаги {steps_total} (база {activity.get('steps_baseline', 0)}), "
f"бонус +{activity.get('total_bonus_kcal', 0)} ккал"
f"TDEE за день: BMR {breakdown.get('bmr')} + NEAT {breakdown.get('neat_kcal')} + "
f"шаги {breakdown.get('steps_kcal')} ({steps_total} шаг.) + "
f"тренировки {breakdown.get('workout_kcal')} = {breakdown.get('tdee')} ккал → "
f"цель {breakdown.get('calorie_target')} ккал"
)
elif steps_total == 0:
lines.append(
"Шаги/тренировки не внесены — TDEE считается как BMR + NEAT. "
"log_steps / log_workout для точной дневной цели."
)
base_cal = targets_base.get("calories", profile.get("calorie_target"))
lines.append(f"Эффективная цель ккал: {base_cal} → {targets.get('calories', base_cal)}")
lines.append("")
lines.append(
@@ -61,7 +70,7 @@ def format_fitness_context(snapshot: dict[str, Any]) -> str:
if stats.get("count"):
lines.append(
f"Тренировки за {stats.get('days', 7)} дн.: {stats.get('count')} "
f"(цель/нед {stats.get('weekly_target')}, серия {stats.get('streak')} дн.)"
f"(серия {stats.get('streak')} дн., {stats.get('active_kcal')} ккал активных)"
)
latest = (snapshot.get("body_metrics") or [None])[0]
@@ -89,6 +98,9 @@ def format_fitness_context(snapshot: dict[str, Any]) -> str:
"Правила: log_meal, log_water, log_weight (обхваты → Navy), log_steps, log_workout (date/days_ago), "
"calc_body_composition (расчёт без записи), get_fitness_summary (date/days_ago), get_fitness_history, "
"set_fitness_profile, calc_fitness_targets, lookup_food, lookup_exercise. "
"Еда — оценка LLM (≈), пользователь может уточнить."
"TDEE = BMR + NEAT (200 ккал) + шаги + тренировки. "
"БЖУ: белок 2.2 г/кг (сушка) / 1.8 г/кг (поддержание/набор), жир 1.0 г/кг, угли — остаток от целевых ккал. "
"Скриншоты Mi Fitness: vision уже извлекла данные в блок [Скриншот] с fitness_hints — используй их, не говори что не видишь картинку. "
"Еда — оценка LLM (≈)."
)
return chr(10).join(lines)
backend/app/fitness/service.py
+59 -56
View File
@@ -14,13 +14,14 @@ from app.db.models import (
WaterLog,
WorkoutLog,
)
from app.fitness.activity_budget import (
build_base_targets,
compute_activity_bonus,
estimate_workout_active_kcal,
scale_targets,
from app.fitness.activity_budget import estimate_workout_active_kcal
from app.fitness.calculators import (
compute_daily_targets,
compute_targets,
one_rep_max,
targets_to_api,
tdee_breakdown_to_api,
)
from app.fitness.calculators import compute_targets, one_rep_max
from app.fitness.body_composition import compute_body_composition
DEFAULT_REMINDERS = [
@@ -45,28 +46,26 @@ class FitnessService:
return None
return self._profile_to_dict(row)
def _profile_to_dict(self, row: FitnessProfile) -> dict[str, Any]:
targets = compute_targets(
{
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"activity_level": row.activity_level,
"goal": row.goal,
}
)
def _profile_params(self, row: FitnessProfile) -> dict[str, Any]:
return {
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"goal": row.goal,
"neat_base_kcal": row.neat_base_kcal,
}
def _profile_to_dict(self, row: FitnessProfile) -> dict[str, Any]:
targets = compute_targets(self._profile_params(row))
return {
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"activity_level": row.activity_level,
"goal": row.goal,
"target_weight_kg": row.target_weight_kg,
"weekly_workouts": row.weekly_workouts,
"baseline_steps": row.baseline_steps,
"baseline_workout_kcal": row.baseline_workout_kcal,
"neat_base_kcal": row.neat_base_kcal,
"calorie_target": row.calorie_target,
"protein_g": row.protein_g,
"fat_g": row.fat_g,
@@ -85,23 +84,13 @@ class FitnessService:
self.db.flush()
for key in (
"sex", "age", "height_cm", "weight_kg", "activity_level",
"goal", "target_weight_kg", "weekly_workouts",
"baseline_steps", "baseline_workout_kcal",
"sex", "age", "height_cm", "weight_kg",
"goal", "target_weight_kg", "neat_base_kcal",
):
if key in updates and updates[key] is not None:
setattr(row, key, updates[key])
targets = compute_targets(
{
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"activity_level": row.activity_level,
"goal": row.goal,
}
)
targets = compute_targets(self._profile_params(row))
row.calorie_target = targets["calorie_target"]
row.protein_g = targets["protein_g"]
row.fat_g = targets["fat_g"]
@@ -193,14 +182,12 @@ class FitnessService:
if profile:
return profile
return {
"calorie_target": 2000,
"protein_g": 140,
"fat_g": 65,
"carbs_g": 200,
"water_l": 2.5,
"weight_kg": 70,
"activity_level": "moderate",
"weekly_workouts": 3,
"height_cm": 170,
"age": 30,
"sex": "male",
"goal": "maintain",
"neat_base_kcal": 200,
}
@@ -248,24 +235,19 @@ class FitnessService:
"steps": steps_total,
}
base_targets = build_base_targets(profile)
activity = compute_activity_bonus(
daily = compute_daily_targets(
profile,
steps_total=steps_total,
workouts=workouts,
)
effective_targets, targets_base = scale_targets(
base_targets,
activity.total_bonus_kcal,
)
targets = targets_to_api(daily)
return {
"date": (day or datetime.now(timezone.utc).date()).isoformat(),
"profile_configured": profile_row is not None,
"totals": totals,
"targets": effective_targets,
"targets_base": targets_base,
"activity": activity.to_dict(),
"targets": targets,
"tdee_breakdown": tdee_breakdown_to_api(daily),
"meals": [self._food_to_dict(f) for f in foods],
"water": [self._water_to_dict(w) for w in waters],
"workouts": workouts,
@@ -393,8 +375,8 @@ class FitnessService:
"age": profile_row.age,
"height_cm": profile_row.height_cm,
"weight_kg": weight_kg,
"activity_level": profile_row.activity_level,
"goal": profile_row.goal,
"neat_base_kcal": profile_row.neat_base_kcal,
}
)
profile_row.calorie_target = targets["calorie_target"]
@@ -428,10 +410,27 @@ class FitnessService:
active_calories: float | None = None,
total_calories: float | None = None,
steps: int | None = None,
activity_type: str | None = None,
met: float | None = None,
logged_at: datetime | str | None = None,
day: date | None = None,
days_ago: int | None = None,
) -> dict[str, Any]:
profile = self.get_profile() or {}
weight_kg = float(profile.get("weight_kg") or 70)
if active_calories is None and duration_min and met is not None:
active_calories = round(met * weight_kg * (float(duration_min) / 60.0), 1)
elif active_calories is None and duration_min:
draft = {
"title": title,
"notes": notes,
"activity_type": activity_type,
"met": met,
"duration_min": duration_min,
}
active_calories = estimate_workout_active_kcal(draft, weight_kg=weight_kg) or None
row = WorkoutLog(
user_id=self.user_id,
title=title[:255],
@@ -471,12 +470,16 @@ class FitnessService:
).all()
profile = self.get_profile() or {}
weekly_target = int(profile.get("weekly_workouts") or 3)
weight_kg = float(profile.get("weight_kg") or 70)
weekly_target = 3
count = len(rows)
duration_min = sum(r.duration_min or 0 for r in rows)
active_kcal = round(
sum(estimate_workout_active_kcal(self._workout_to_dict(r)) for r in rows),
sum(
estimate_workout_active_kcal(self._workout_to_dict(r), weight_kg=weight_kg)
for r in rows
),
1,
)
@@ -583,7 +586,7 @@ class FitnessService:
*,
days: int = 7,
end_day: date | None = None,
include_targets_base: bool = True,
include_tdee_breakdown: bool = True,
) -> dict[str, Any]:
days = max(1, min(days, 90))
end = end_day or datetime.now(timezone.utc).date()
@@ -603,8 +606,8 @@ class FitnessService:
"meal_count": len(full["meals"]),
"workout_count": len(full["workouts"]),
}
if include_targets_base:
item["targets_base"] = full.get("targets_base")
if include_tdee_breakdown:
item["tdee_breakdown"] = full.get("tdee_breakdown")
summaries.append(item)
return {
backend/app/fitness/structuring.py
+7 -1
View File
@@ -28,8 +28,10 @@ WORKOUT_PROMPT = """
Формат:
{
"title": "название",
"activity_type": "ходьба|бег|силовая|велосипед|плавание|йога|hiit|другое",
"duration_min": null,
"active_calories": null,
"met": null,
"total_calories": null,
"steps": null,
"notes": "",
@@ -39,7 +41,11 @@ WORKOUT_PROMPT = """
}
Правила:
- weight_kg в кг, округляй разумно.
- active_calories / total_calories / steps — если упомянуты в тексте, иначе null.
- active_calories — только если явно указаны в тексте, иначе null.
- duration_min — длительность в минутах, если можно оценить из текста.
- met — MET по Compendium of Physical Activities, если ккал не указаны (ходьба ~3.5, бег ~9.8, силовая ~6, велосипед ~7.5, плавание ~8, йога ~3, hiit ~8).
- activity_type — тип активности для расчёта MET.
- total_calories / steps — если упомянуты в тексте, иначе null.
- Если данных нет — null или пустой массив.
""".strip()
backend/app/homelab/digest.py
+9 -5
View File
@@ -7,7 +7,7 @@ from app.homelab.rss import RssClient
def build_morning_digest(db: Session, *, include_news: bool = True) -> str:
del db # timezone resolved via weather client / profile in future extensions
weather_client = OpenMeteoClient()
weather = weather_client.fetch_current_and_hourly(hours_ahead=12)
weather = weather_client.fetch_forecast(hours_ahead=12, days_ahead=7)
lines = ["🌤 **Утренний дайджест**", ""]
@@ -18,7 +18,10 @@ def build_morning_digest(db: Session, *, include_news: bool = True) -> str:
f"{cur.get('temperature_c')}°C, {cur.get('conditions')}, "
f"ветер {cur.get('wind_speed_kmh')} км/ч."
)
lines.append(weather_client.rain_summary(hours_ahead=12))
lines.append(weather_client.rain_summary(hours_ahead=12, daily=weather.get("daily")))
daily = weather_client.daily_summary(days_ahead=7)
if daily:
lines.append(f"**На неделю**: {daily}")
else:
lines.append(f"**Погода**: недоступна ({weather.get('error', 'ошибка')}).")
@@ -41,12 +44,13 @@ def build_morning_digest(db: Session, *, include_news: bool = True) -> str:
return "\n".join(lines)
def build_weather_briefing(hours_ahead: int = 12, include_news: bool = False) -> dict:
def build_weather_briefing(hours_ahead: int = 12, days_ahead: int = 7, include_news: bool = False) -> dict:
client = OpenMeteoClient()
weather = client.fetch_current_and_hourly(hours_ahead=hours_ahead)
weather = client.fetch_forecast(hours_ahead=hours_ahead, days_ahead=days_ahead)
result = {
"weather": weather,
"rain_summary": client.rain_summary(hours_ahead=hours_ahead) if weather.get("ok") else "",
"rain_summary": client.rain_summary(hours_ahead=hours_ahead, daily=weather.get("daily")) if weather.get("ok") else "",
"daily_summary": client.daily_summary(days_ahead=days_ahead) if weather.get("ok") else "",
"context": format_weather_snapshot(weather),
}
if include_news:
backend/app/homelab/openmeteo.py
+248 -59
View File
@@ -29,12 +29,19 @@ WEATHER_CODES: dict[int, str] = {
99: "гроза с градом",
}
WEATHER_QUERY_KEYWORDS = (
"погод", "дожд", "снег", "ветер", "температур", "градус", "мороз", "жар",
"на улице", "одеть", "зонт", "прогноз", "завтра", "послезавтра", "выходн",
"weather", "rain", "forecast", "umbrella", "outside",
)
_cache: dict[str, Any] = {
"data": None,
"fetched_at": 0.0,
"expires_at": 0.0,
"source": "local",
"local_coverage": {"current": [], "hourly": []},
"local_coverage": {"current": [], "hourly": [], "daily": []},
"merged_fields": [],
}
CURRENT_FIELDS = (
@@ -51,6 +58,14 @@ HOURLY_FIELDS = (
"precipitation",
"weather_code",
)
DAILY_FIELDS = (
"weather_code",
"temperature_2m_max",
"temperature_2m_min",
"precipitation_sum",
"precipitation_probability_max",
"wind_speed_10m_max",
)
RECOMMENDED_SYNC_DOMAINS = "dwd_icon,ncep_gfs013,ncep_gefs025"
RECOMMENDED_SYNC_VARIABLES = (
@@ -58,11 +73,20 @@ RECOMMENDED_SYNC_VARIABLES = (
"precipitation,rain,cloud_cover,weather_code,wind_u_component_10m,wind_v_component_10m"
)
SYNC_HINT = (
"Контейнер open-meteo-sync, скорее всего, качает только temperature_2m. "
f"Задай SYNC_DOMAINS={RECOMMENDED_SYNC_DOMAINS} и "
"Локальный open-meteo-sync отдаёт неполные данные. "
f"SYNC_DOMAINS={RECOMMENDED_SYNC_DOMAINS} "
f"SYNC_VARIABLES={RECOMMENDED_SYNC_VARIABLES} (~12 GB). "
"Документация: github.com/open-meteo/open-data/tree/main/tutorial_weather_api"
)
PRECIP_PROB_HINT = (
"Для вероятности дождя добавь ncep_gefs025 в SYNC_DOMAINS "
"и precipitation_probability в SYNC_VARIABLES."
)
def weather_query_relevant(query: str) -> bool:
q = (query or "").lower()
return any(kw in q for kw in WEATHER_QUERY_KEYWORDS)
def _hourly_series(hourly: dict[str, Any], key: str) -> list[Any]:
@@ -70,6 +94,11 @@ def _hourly_series(hourly: dict[str, Any], key: str) -> list[Any]:
return values if isinstance(values, list) else []
def _daily_series(daily: dict[str, Any], key: str) -> list[Any]:
values = daily.get(key)
return values if isinstance(values, list) else []
def _hourly_start_index(times: list[str], anchor_time: str | None) -> int:
if not times:
return 0
@@ -85,18 +114,21 @@ def _hourly_start_index(times: list[str], anchor_time: str | None) -> int:
def _field_coverage(raw: dict[str, Any]) -> dict[str, list[str]]:
"""Какие поля реально пришли от OpenMeteo (не null)."""
current = raw.get("current") or {}
hourly = raw.get("hourly") or {}
current_present = [
key for key in CURRENT_FIELDS if current.get(key) is not None
]
daily = raw.get("daily") or {}
current_present = [key for key in CURRENT_FIELDS if current.get(key) is not None]
hourly_present = []
for key in HOURLY_FIELDS:
series = _hourly_series(hourly, key)
if any(v is not None for v in series):
hourly_present.append(key)
return {"current": current_present, "hourly": hourly_present}
daily_present = []
for key in DAILY_FIELDS:
series = _daily_series(daily, key)
if any(v is not None for v in series):
daily_present.append(key)
return {"current": current_present, "hourly": hourly_present, "daily": daily_present}
def _coverage_sufficient(coverage: dict[str, list[str]]) -> bool:
@@ -106,11 +138,27 @@ def _coverage_sufficient(coverage: dict[str, list[str]]) -> bool:
return False
if len(current) < 3:
return False
if "precipitation_probability" not in hourly and "weather_code" not in hourly:
if "weather_code" not in hourly and "temperature_2m" not in hourly:
return False
return True
def _local_needs_sync_hint(local_coverage: dict[str, list[str]]) -> bool:
current = set(local_coverage.get("current") or [])
hourly = set(local_coverage.get("hourly") or [])
if "temperature_2m" not in current:
return True
if "weather_code" not in current:
return True
if "temperature_2m" not in hourly:
return True
return False
def _missing_precip_probability(coverage: dict[str, list[str]]) -> bool:
return "precipitation_probability" not in set(coverage.get("hourly") or [])
def _fmt_num(value: Any, *, suffix: str = "") -> str:
if value is None:
return "—"
@@ -121,6 +169,42 @@ def _fmt_num(value: Any, *, suffix: str = "") -> str:
return f"{text}{suffix}" if suffix else text
def _conditions(code: Any) -> str:
if code is None:
return "неизвестно"
return WEATHER_CODES.get(int(code), "неизвестно")
def _format_day_label(date_str: str, index: int) -> str:
if index == 0:
return "Сегодня"
if index == 1:
return "Завтра"
if not date_str:
return f"День {index + 1}"
parts = date_str.split("-")
if len(parts) == 3:
return f"{parts[2]}.{parts[1]}"
return date_str
def _merge_hourly_field(target: dict[str, Any], source: dict[str, Any], field: str) -> bool:
hourly_t = target.setdefault("hourly", {})
hourly_s = source.get("hourly") or {}
src = hourly_s.get(field)
if not isinstance(src, list) or not any(v is not None for v in src):
return False
dst = hourly_t.get(field)
if isinstance(dst, list) and len(dst) == len(src):
hourly_t[field] = [
dst[i] if dst[i] is not None else src[i]
for i in range(len(src))
]
else:
hourly_t[field] = src
return True
class OpenMeteoClient:
def __init__(self) -> None:
settings = get_settings()
@@ -131,6 +215,7 @@ class OpenMeteoClient:
self.lon = settings.weather_lon
self.location_name = settings.weather_location_name
self.cache_ttl = settings.weather_cache_sec
self.forecast_days = max(2, int(settings.weather_forecast_days or 7))
def _request_params(self) -> dict[str, Any]:
return {
@@ -138,8 +223,9 @@ class OpenMeteoClient:
"longitude": self.lon,
"current": ",".join(CURRENT_FIELDS),
"hourly": ",".join(HOURLY_FIELDS),
"daily": ",".join(DAILY_FIELDS),
"timezone": "auto",
"forecast_days": 2,
"forecast_days": self.forecast_days,
}
def _fetch_from_url(self, base_url: str) -> dict[str, Any]:
@@ -157,18 +243,26 @@ class OpenMeteoClient:
local_coverage = _field_coverage(local_raw)
source = "local"
raw = local_raw
merged_fields: list[str] = []
if (
need_fallback = (
self.fallback_on_partial
and self.fallback_url
and self.fallback_url.rstrip("/") != self.base_url
and not _coverage_sufficient(local_coverage)
):
)
if need_fallback:
try:
fallback_raw = self._fetch_from_url(self.fallback_url)
if _coverage_sufficient(_field_coverage(fallback_raw)):
fallback_coverage = _field_coverage(fallback_raw)
if not _coverage_sufficient(local_coverage) and _coverage_sufficient(fallback_coverage):
raw = fallback_raw
source = "fallback"
elif _missing_precip_probability(local_coverage) and not _missing_precip_probability(fallback_coverage):
if _merge_hourly_field(raw, fallback_raw, "precipitation_probability"):
merged_fields.append("precipitation_probability")
source = "merged"
except Exception:
pass
@@ -177,6 +271,7 @@ class OpenMeteoClient:
_cache["expires_at"] = now + self.cache_ttl
_cache["source"] = source
_cache["local_coverage"] = local_coverage
_cache["merged_fields"] = merged_fields
return raw
def cache_status(self) -> dict[str, Any]:
@@ -194,43 +289,78 @@ class OpenMeteoClient:
"ttl_sec": self.cache_ttl,
"expires_in_sec": expires_in_sec,
"source": _cache.get("source") or "local",
"merged_fields": list(_cache.get("merged_fields") or []),
}
def fetch_current_and_hourly(self, hours_ahead: int = 12) -> dict[str, Any]:
def _build_hourly_slice(self, raw: dict[str, Any], hours_ahead: int) -> list[dict[str, Any]]:
current = raw.get("current") or {}
hourly = raw.get("hourly") or {}
times = hourly.get("time") or []
start = _hourly_start_index(times, current.get("time"))
end = min(start + hours_ahead, len(times))
rows: list[dict[str, Any]] = []
for i in range(start, end):
code = _hourly_series(hourly, "weather_code")[i] if i < len(_hourly_series(hourly, "weather_code")) else None
temp_series = _hourly_series(hourly, "temperature_2m")
precip_series = _hourly_series(hourly, "precipitation")
prob_series = _hourly_series(hourly, "precipitation_probability")
rows.append({
"time": times[i],
"temperature_c": temp_series[i] if i < len(temp_series) else None,
"precipitation_mm": precip_series[i] if i < len(precip_series) else None,
"precipitation_probability": prob_series[i] if i < len(prob_series) else None,
"weather_code": code,
"conditions": _conditions(code),
})
return rows
def _build_daily_slice(self, raw: dict[str, Any], days_ahead: int) -> list[dict[str, Any]]:
daily = raw.get("daily") or {}
times = daily.get("time") or []
limit = min(days_ahead, len(times))
rows: list[dict[str, Any]] = []
for i in range(limit):
code = _daily_series(daily, "weather_code")[i] if i < len(_daily_series(daily, "weather_code")) else None
rows.append({
"date": times[i],
"label": _format_day_label(times[i], i),
"temperature_max_c": _daily_series(daily, "temperature_2m_max")[i] if i < len(_daily_series(daily, "temperature_2m_max")) else None,
"temperature_min_c": _daily_series(daily, "temperature_2m_min")[i] if i < len(_daily_series(daily, "temperature_2m_min")) else None,
"precipitation_sum_mm": _daily_series(daily, "precipitation_sum")[i] if i < len(_daily_series(daily, "precipitation_sum")) else None,
"precipitation_probability_max": _daily_series(daily, "precipitation_probability_max")[i] if i < len(_daily_series(daily, "precipitation_probability_max")) else None,
"wind_speed_max_kmh": _daily_series(daily, "wind_speed_10m_max")[i] if i < len(_daily_series(daily, "wind_speed_10m_max")) else None,
"weather_code": code,
"conditions": _conditions(code),
})
return rows
def fetch_forecast(self, hours_ahead: int = 12, days_ahead: int = 7) -> dict[str, Any]:
hours_ahead = max(1, min(int(hours_ahead), 168))
days_ahead = max(1, min(int(days_ahead), self.forecast_days))
try:
raw = self._fetch_raw()
except Exception as exc:
return {"ok": False, "error": str(exc), "location": self.location_name}
current = raw.get("current") or {}
hourly = raw.get("hourly") or {}
times = hourly.get("time") or []
start = _hourly_start_index(times, current.get("time"))
end = min(start + hours_ahead, len(times))
hourly_slice = []
for i in range(start, end):
code = _hourly_series(hourly, "weather_code")[i] if i < len(_hourly_series(hourly, "weather_code")) else None
temp_series = _hourly_series(hourly, "temperature_2m")
precip_series = _hourly_series(hourly, "precipitation")
prob_series = _hourly_series(hourly, "precipitation_probability")
hourly_slice.append({
"time": times[i],
"temperature_c": temp_series[i] if i < len(temp_series) else None,
"precipitation_mm": precip_series[i] if i < len(precip_series) else None,
"precipitation_probability": prob_series[i] if i < len(prob_series) else None,
"weather_code": code,
"conditions": WEATHER_CODES.get(code, "неизвестно") if code is not None else "неизвестно",
})
code = current.get("weather_code")
coverage = _field_coverage(raw)
local_coverage = _cache.get("local_coverage") or coverage
sync_hint = ""
if _local_needs_sync_hint(local_coverage):
sync_hint = SYNC_HINT
elif _missing_precip_probability(local_coverage):
sync_hint = PRECIP_PROB_HINT
return {
"ok": True,
"location": self.location_name,
"data_source": _cache.get("source") or "local",
"local_field_coverage": _cache.get("local_coverage") or coverage,
"merged_fields": list(_cache.get("merged_fields") or []),
"local_field_coverage": local_coverage,
"field_coverage": coverage,
"sync_hint": SYNC_HINT if not _coverage_sufficient(_cache.get("local_coverage") or coverage) else "",
"sync_hint": sync_hint,
"current": {
"time": current.get("time"),
"temperature_c": current.get("temperature_2m"),
@@ -239,13 +369,17 @@ class OpenMeteoClient:
"precipitation_mm": current.get("precipitation"),
"wind_speed_kmh": current.get("wind_speed_10m"),
"weather_code": code,
"conditions": WEATHER_CODES.get(code, "неизвестно") if code is not None else "неизвестно",
"conditions": _conditions(code),
},
"hourly": hourly_slice,
"hourly": self._build_hourly_slice(raw, hours_ahead),
"daily": self._build_daily_slice(raw, days_ahead),
}
def rain_summary(self, hours_ahead: int = 12) -> str:
data = self.fetch_current_and_hourly(hours_ahead=hours_ahead)
def fetch_current_and_hourly(self, hours_ahead: int = 12) -> dict[str, Any]:
return self.fetch_forecast(hours_ahead=hours_ahead, days_ahead=min(7, self.forecast_days))
def rain_summary(self, hours_ahead: int = 12, daily: list[dict[str, Any]] | None = None) -> str:
data = self.fetch_forecast(hours_ahead=hours_ahead, days_ahead=2)
if not data.get("ok"):
return f"Погода недоступна: {data.get('error', 'ошибка')}"
@@ -255,16 +389,49 @@ class OpenMeteoClient:
precip = hour.get("precipitation_mm") or 0
if (prob is not None and prob >= 40) or precip > 0:
time_str = (hour.get("time") or "")[11:16]
rainy_hours.append(f"{time_str} ({prob}% вероятность, {precip} мм)")
prob_text = f"{prob}%" if prob is not None else "—"
rainy_hours.append(f"{time_str} ({prob_text}, {precip} мм)")
lines: list[str] = []
if rainy_hours:
return "Ожидаются осадки: " + ", ".join(rainy_hours[:6])
return "Существенных осадков в ближайшие часы не ожидается."
lines.append("Ожидаются осадки: " + ", ".join(rainy_hours[:6]))
else:
lines.append("Существенных осадков в ближайшие часы не ожидается.")
days = daily if daily is not None else data.get("daily") or []
if len(days) > 1:
tomorrow = days[1]
tmax = tomorrow.get("temperature_max_c")
tmin = tomorrow.get("temperature_min_c")
prob = tomorrow.get("precipitation_probability_max")
precip = tomorrow.get("precipitation_sum_mm") or 0
cond = tomorrow.get("conditions") or "неизвестно"
prob_part = f", дождь до {prob}%" if prob is not None and prob >= 30 else ""
precip_part = f", {precip} мм" if precip > 0 else ""
lines.append(
f"Завтра: {_fmt_num(tmin)}–{_fmt_num(tmax, suffix='°C')}, {cond}{prob_part}{precip_part}."
)
return " ".join(lines)
def daily_summary(self, days_ahead: int = 7) -> str:
data = self.fetch_forecast(hours_ahead=1, days_ahead=days_ahead)
if not data.get("ok"):
return ""
parts = []
for day in data.get("daily") or []:
label = day.get("label") or day.get("date")
tmax = day.get("temperature_max_c")
tmin = day.get("temperature_min_c")
cond = day.get("conditions") or "неизвестно"
prob = day.get("precipitation_probability_max")
prob_part = f", дождь до {prob}%" if prob is not None and prob >= 30 else ""
parts.append(f"{label}: {_fmt_num(tmin)}–{_fmt_num(tmax, suffix='°C')}, {cond}{prob_part}")
return "; ".join(parts)
def format_weather_snapshot(data: dict[str, Any] | None = None) -> str:
def format_weather_snapshot(data: dict[str, Any] | None = None, *, include_daily: bool = True) -> str:
client = OpenMeteoClient()
snapshot = data if data is not None else client.fetch_current_and_hourly(hours_ahead=6)
snapshot = data if data is not None else client.fetch_forecast(hours_ahead=6, days_ahead=3)
lines = ["[Погода]"]
if not snapshot.get("ok"):
@@ -281,30 +448,50 @@ def format_weather_snapshot(data: dict[str, Any] | None = None) -> str:
f"{snapshot.get('location')}: {_fmt_num(cur.get('temperature_c'), suffix='°C')}"
f"{apparent_part}, {cur.get('conditions') or 'неизвестно'}{wind_part}."
)
hourly = snapshot.get("hourly") or []
rainy_hours = []
for hour in hourly:
for hour in snapshot.get("hourly") or []:
prob = hour.get("precipitation_probability")
precip = hour.get("precipitation_mm") or 0
if (prob is not None and prob >= 40) or precip > 0:
time_str = (hour.get("time") or "")[11:16]
rainy_hours.append(f"{time_str} ({prob}% вероятность, {precip} мм)")
prob_text = f"{prob}%" if prob is not None else "—"
rainy_hours.append(f"{time_str} ({prob_text}, {precip} мм)")
if rainy_hours:
lines.append("Ожидаются осадки: " + ", ".join(rainy_hours[:6]))
else:
lines.append("Существенных осадков в ближайшие часы не ожидается.")
lines.append("Вопросы «что на улице» / «будет ли дождь» — get_weather.")
if include_daily:
days = snapshot.get("daily") or []
if len(days) > 1:
tomorrow = days[1]
lines.append(
f"Завтра: {_fmt_num(tomorrow.get('temperature_min_c'))}–"
f"{_fmt_num(tomorrow.get('temperature_max_c'), suffix='°C')}, "
f"{tomorrow.get('conditions') or 'неизвестно'}."
)
if len(days) > 2:
week_bits = []
for day in days[2:7]:
week_bits.append(
f"{day.get('label')}: {_fmt_num(day.get('temperature_min_c'))}–"
f"{_fmt_num(day.get('temperature_max_c'), suffix='°C')}"
)
if week_bits:
lines.append("Далее: " + "; ".join(week_bits) + ".")
lines.append("Подробнее — get_weather (hours_ahead, days_ahead).")
return "\n".join(lines)
def build_weather_dashboard(hours_ahead: int = 12) -> dict[str, Any]:
"""Полный снимок для UI: данные OpenMeteo + контекст ассистента."""
def build_weather_dashboard(hours_ahead: int = 12, days_ahead: int = 7) -> dict[str, Any]:
client = OpenMeteoClient()
weather = client.fetch_current_and_hourly(hours_ahead=hours_ahead)
settings = get_settings()
weather = client.fetch_forecast(hours_ahead=hours_ahead, days_ahead=days_ahead)
return {
"weather": weather,
"rain_summary": client.rain_summary(hours_ahead=hours_ahead) if weather.get("ok") else "",
"rain_summary": client.rain_summary(hours_ahead=hours_ahead, daily=weather.get("daily")) if weather.get("ok") else "",
"daily_summary": client.daily_summary(days_ahead=days_ahead) if weather.get("ok") else "",
"assistant_context": format_weather_snapshot(weather),
"cache": client.cache_status(),
"config": {
@@ -313,24 +500,26 @@ def build_weather_dashboard(hours_ahead: int = 12) -> dict[str, Any]:
"longitude": client.lon,
"openmeteo_base_url": client.base_url,
"cache_ttl_sec": client.cache_ttl,
"forecast_days": 2,
"forecast_days": client.forecast_days,
"timezone": "auto",
},
"available_fields": {
"current": list(CURRENT_FIELDS),
"hourly": list(HOURLY_FIELDS),
"daily": list(DAILY_FIELDS),
},
"field_coverage": weather.get("field_coverage") if weather.get("ok") else {"current": [], "hourly": []},
"local_field_coverage": weather.get("local_field_coverage") if weather.get("ok") else {"current": [], "hourly": []},
"field_coverage": weather.get("field_coverage") if weather.get("ok") else {"current": [], "hourly": [], "daily": []},
"local_field_coverage": weather.get("local_field_coverage") if weather.get("ok") else {"current": [], "hourly": [], "daily": []},
"data_source": weather.get("data_source", "local") if weather.get("ok") else "local",
"merged_fields": weather.get("merged_fields", []) if weather.get("ok") else [],
"sync_hint": weather.get("sync_hint", "") if weather.get("ok") else SYNC_HINT,
"recommended_sync": {
"domains": RECOMMENDED_SYNC_DOMAINS,
"variables": RECOMMENDED_SYNC_VARIABLES,
},
"assistant_tools": {
"get_weather": "Текущая погода и почасовой прогнос (hours_ahead до 48)",
"get_weather": "Сейчас + почасово (hours_ahead до 168) + по дням (days_ahead до 16)",
"get_morning_briefing": "Погода + заголовки RSS-новостей",
},
"system_prompt": "Краткий блок [Погода] в system prompt каждого сообщения (6 ч почасово).",
"system_prompt": "Блок [Погода] в system prompt — только если запрос про погоду/одежду/прогноз.",
}
backend/app/llm/client.py
+51
View File
@@ -34,6 +34,16 @@ class LLMClient:
finally:
db.close()
def _vision_model_runtime(self) -> str:
from app.db.base import SessionLocal
from app.settings.service import SettingsService
db = SessionLocal()
try:
return str(SettingsService(db).get_effective("openrouter_vision_model"))
finally:
db.close()
@property
def model(self) -> str:
return self._runtime()[0]
@@ -46,6 +56,10 @@ class LLMClient:
def reasoning_effort(self) -> str:
return self._runtime()[2]
@property
def vision_model(self) -> str:
return self._vision_model_runtime()
def _reasoning_extra_body(self) -> dict[str, Any] | None:
if not self.reasoning_effort:
return None
@@ -272,6 +286,43 @@ class LLMClient:
return result
async def complete_vision(
self,
messages: list[dict[str, Any]],
*,
temperature: float = 0.1,
model: str | None = None,
) -> dict[str, Any]:
use_model = model or self.vision_model
kwargs: dict[str, Any] = {
"model": use_model,
"messages": messages,
"temperature": temperature,
"extra_body": {"reasoning": {"effort": "none", "exclude": True}},
}
response = await self.client.chat.completions.create(**kwargs)
usage = getattr(response, "usage", None)
usage_dict: dict[str, Any] = {}
if usage is not None:
usage_dict = {
"prompt_tokens": getattr(usage, "prompt_tokens", None),
"completion_tokens": getattr(usage, "completion_tokens", None),
"total_tokens": getattr(usage, "total_tokens", None),
}
logger.info(
"LLM vision usage: prompt=%s completion=%s total=%s model=%s",
usage_dict.get("prompt_tokens"),
usage_dict.get("completion_tokens"),
usage_dict.get("total_tokens"),
use_model,
)
message = response.choices[0].message
return {
"content": message.content or "",
"model": use_model,
"usage": usage_dict,
}
@staticmethod
def parse_tool_arguments(arguments: str) -> dict[str, Any]:
if not arguments:
backend/app/settings/service.py
+5 -1
View File
@@ -7,12 +7,13 @@ from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import Settings, get_settings
from app.config import Settings, get_settings, resolve_vision_model
from app.db.models import AssistantState
SETTING_KEYS = (
"openrouter_model",
"memory_extract_model",
"openrouter_vision_model",
"openrouter_reasoning_effort",
"rag_enabled",
"rag_top_k",
@@ -48,6 +49,7 @@ class SettingsService:
mapping = {
"openrouter_model": defaults.openrouter_model,
"memory_extract_model": defaults.memory_extract_model or defaults.openrouter_model,
"openrouter_vision_model": defaults.openrouter_vision_model,
"openrouter_reasoning_effort": defaults.openrouter_reasoning_effort,
"rag_enabled": defaults.rag_enabled,
"rag_top_k": defaults.rag_top_k,
@@ -65,6 +67,8 @@ class SettingsService:
return max(1, min(50, int(raw)))
except ValueError:
return self._default_for(key)
if key == "openrouter_vision_model":
return resolve_vision_model(raw.strip())
return raw
def snapshot(self) -> dict[str, Any]:
backend/app/tools/registry.py
+28 -17
View File
@@ -336,7 +336,7 @@ TOOL_DEFINITIONS: list[dict[str, Any]] = [
"type": "function",
"function": {
"name": "set_fitness_profile",
"description": "Настроить фитнес-профиль и пересчитать цели ккал/БЖУ/воды.",
"description": "Настроить фитнес-профиль и пересчитать цели ккал/БЖУ/воды (TDEE = BMR + NEAT).",
"parameters": {
"type": "object",
"properties": {
@@ -344,13 +344,12 @@ TOOL_DEFINITIONS: list[dict[str, Any]] = [
"age": {"type": "integer"},
"height_cm": {"type": "number"},
"weight_kg": {"type": "number"},
"activity_level": {
"type": "string",
"description": "sedentary/light/moderate/active/very_active",
},
"goal": {"type": "string", "description": "lose/maintain/gain"},
"target_weight_kg": {"type": "number"},
"weekly_workouts": {"type": "integer"},
"neat_base_kcal": {
"type": "number",
"description": "NEAT-база 200–300 ккал, по умолчанию 200",
},
},
"required": [],
},
@@ -360,7 +359,7 @@ TOOL_DEFINITIONS: list[dict[str, Any]] = [
"type": "function",
"function": {
"name": "calc_fitness_targets",
"description": "Калькулятор BMR/TDEE/макросов без сохранения.",
"description": "Калькулятор BMR/TDEE/макросов без сохранения (rest-day: BMR + NEAT).",
"parameters": {
"type": "object",
"properties": {
@@ -368,8 +367,9 @@ TOOL_DEFINITIONS: list[dict[str, Any]] = [
"age": {"type": "integer"},
"height_cm": {"type": "number"},
"weight_kg": {"type": "number"},
"activity_level": {"type": "string"},
"goal": {"type": "string"},
"neat_base_kcal": {"type": "number"},
"steps": {"type": "integer", "description": "Шаги за день для расчёта TDEE"},
},
"required": ["weight_kg", "height_cm", "age"],
},
@@ -539,15 +539,19 @@ TOOL_DEFINITIONS: list[dict[str, Any]] = [
"function": {
"name": "get_weather",
"description": (
"ОБЯЗАТЕЛЬНО для вопросов о погоде, «что на улице», «будет ли дождь». "
"Текущая погода и прогноз по часам."
"ОБЯЗАТЕЛЬНО для вопросов о погоде, «что на улице», «будет ли дождь», «завтра», «на неделю». "
"Текущая погода, почасовой и дневной прогноз."
),
"parameters": {
"type": "object",
"properties": {
"hours_ahead": {
"type": "integer",
"description": "Сколько часов прогноза (по умолчанию 12)",
"description": "Сколько часов почасового прогноза (по умолчанию 12, до 168)",
},
"days_ahead": {
"type": "integer",
"description": "Сколько дней дневного прогноза (по умолчанию 7, до 16)",
},
},
"required": [],
@@ -917,14 +921,17 @@ async def execute_tool(
updates = {
k: arguments[k]
for k in (
"sex", "age", "height_cm", "weight_kg", "activity_level",
"goal", "target_weight_kg", "weekly_workouts",
"sex", "age", "height_cm", "weight_kg",
"goal", "target_weight_kg", "neat_base_kcal",
)
if k in arguments and arguments[k] is not None
}
result = fitness.set_profile(updates)
elif name == "calc_fitness_targets":
result = fitness.calc_targets(arguments)
from app.fitness.calculators import compute_daily_targets
steps = int(arguments.get("steps") or 0)
result = compute_daily_targets(arguments, steps_total=steps, workouts=[])
elif name == "calc_body_composition":
result = fitness.calc_body_composition(arguments)
elif name == "log_meal":
@@ -980,6 +987,8 @@ async def execute_tool(
active_calories=structured.get("active_calories"),
total_calories=structured.get("total_calories"),
steps=structured.get("steps"),
activity_type=structured.get("activity_type"),
met=structured.get("met"),
day=day,
days_ago=arguments.get("days_ago"),
)
@@ -1002,12 +1011,14 @@ async def execute_tool(
interval_hours=arguments.get("interval_hours"),
)
elif name == "get_weather":
hours = int(arguments.get("hours_ahead") or 12)
hours = max(1, min(int(arguments.get("hours_ahead") or 12), 168))
days = max(1, min(int(arguments.get("days_ahead") or 7), 16))
client = OpenMeteoClient()
weather = client.fetch_current_and_hourly(hours_ahead=hours)
weather = client.fetch_forecast(hours_ahead=hours, days_ahead=days)
result = {
"weather": weather,
"rain_summary": client.rain_summary(hours_ahead=hours) if weather.get("ok") else "",
"rain_summary": client.rain_summary(hours_ahead=hours, daily=weather.get("daily")) if weather.get("ok") else "",
"daily_summary": client.daily_summary(days_ahead=days) if weather.get("ok") else "",
}
elif name == "get_morning_briefing":
include_news = arguments.get("include_news", True)
backend/app/vision/__init__.py
+10
View File
@@ -0,0 +1,10 @@
from app.vision.analyze import VisionResult, VisionService, format_user_message, format_user_messages, vision_debug_payload, vision_debug_payloads
__all__ = [
"VisionResult",
"VisionService",
"format_user_message",
"format_user_messages",
"vision_debug_payload",
"vision_debug_payloads",
]
backend/app/vision/analyze.py
+199
View File
@@ -0,0 +1,199 @@
from __future__ import annotations
import base64
import json
import logging
from dataclasses import dataclass, field
from typing import Any
from openai import APIStatusError
from app.llm.client import LLMClient
from app.projects.structuring import strip_markdown_json
from app.vision.preprocess import PreparedImage, prepare_image
from app.vision.prompts import VISION_SYSTEM_PROMPT
logger = logging.getLogger(__name__)
class VisionUnavailableError(Exception):
"""Vision LLM endpoint missing or unreachable on OpenRouter."""
def __init__(self, model: str, detail: str) -> None:
self.model = model
super().__init__(detail)
@dataclass
class VisionResult:
parsed: dict[str, Any] = field(default_factory=dict)
raw_content: str = ""
model: str = ""
usage: dict[str, Any] = field(default_factory=dict)
image_meta: dict[str, Any] = field(default_factory=dict)
parse_error: str | None = None
class VisionService:
def __init__(self) -> None:
self.llm = LLMClient()
async def analyze(self, image_bytes: bytes, *, user_hint: str = "") -> VisionResult:
prepared = prepare_image(image_bytes)
return await self.analyze_prepared(prepared, user_hint=user_hint)
async def analyze_prepared(self, prepared: PreparedImage, *, user_hint: str = "") -> VisionResult:
b64 = base64.standard_b64encode(prepared.jpeg_bytes).decode("ascii")
hint = f"\n\nПодсказка пользователя: {user_hint.strip()}" if user_hint.strip() else ""
messages: list[dict[str, Any]] = [
{"role": "system", "content": VISION_SYSTEM_PROMPT},
{
"role": "user",
"content": [
{"type": "text", "text": f"Извлеки данные со скриншота.{hint}"},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
},
],
},
]
model = self.llm.vision_model
try:
response = await self.llm.complete_vision(messages)
except APIStatusError as exc:
if exc.status_code == 404:
raise VisionUnavailableError(
model,
f"Vision-модель «{model}» недоступна на OpenRouter. "
"Укажите другую в Settings (например google/gemini-2.5-flash-lite).",
) from exc
raise
raw = (response.get("content") or "").strip()
parsed: dict[str, Any] = {}
parse_error: str | None = None
try:
parsed = json.loads(strip_markdown_json(raw))
if not isinstance(parsed, dict):
parse_error = "Vision response is not a JSON object"
parsed = {}
except json.JSONDecodeError as exc:
parse_error = str(exc)
parsed = {"description": raw[:2000], "document_type": "other", "raw_fallback": True}
return VisionResult(
parsed=parsed,
raw_content=raw,
model=str(response.get("model") or self.llm.vision_model),
usage=dict(response.get("usage") or {}),
image_meta=prepared.to_meta(),
parse_error=parse_error,
)
def _format_screenshot_block(
result: VisionResult,
*,
index: int | None = None,
total: int | None = None,
) -> str:
parsed = result.parsed or {}
doc_type = parsed.get("document_type") or "other"
confidence = parsed.get("confidence") or "unknown"
if index is not None and total is not None and total > 1:
header = f"[Скриншот {index}/{total}: {doc_type}, confidence={confidence}]"
else:
header = f"[Скриншот: {doc_type}, confidence={confidence}]"
lines = [header]
if parsed.get("description"):
lines.append(f"Описание: {parsed['description']}")
extracted = parsed.get("extracted_text") or []
if extracted:
lines.append("Текст с экрана:")
lines.extend(f"- {line}" for line in extracted if str(line).strip())
tables = parsed.get("tables") or []
if tables:
lines.append("Таблицы:")
for table in tables:
title = table.get("title") if isinstance(table, dict) else None
if title:
lines.append(f" [{title}]")
rows = table.get("rows") if isinstance(table, dict) else None
if isinstance(rows, list):
for row in rows:
if isinstance(row, list):
lines.append(" | " + " | ".join(str(cell) for cell in row))
hints = parsed.get("fitness_hints")
if hints:
lines.append(f"Подсказки для фитнеса: {json.dumps(hints, ensure_ascii=False)}")
if result.parse_error:
lines.append(f"(parse_error: {result.parse_error})")
return "\n".join(lines)
def format_user_message(caption: str, result: VisionResult) -> str:
return format_user_messages(caption, [result])
def format_user_messages(caption: str, results: list[VisionResult]) -> str:
if not results:
return caption.strip()
total = len(results)
blocks = [
_format_screenshot_block(result, index=index, total=total)
for index, result in enumerate(results, start=1)
]
text = "\n\n".join(blocks)
if caption.strip():
text = f"{text}\n\nПодпись: {caption.strip()}"
return text
VISION_TURN_HINT = (
"[Скриншоты в этом сообщении]: vision уже извлекла данные с каждой картинки в блоки [Скриншот] ниже. "
"Отвечай по Описанию и извлечённому тексту как по увиденному. "
"Не утверждай, что не видишь изображения, и не предлагай настроить vision API."
)
def format_vision_turn_hint(user_text: str) -> str:
if "[Скриншот" not in (user_text or ""):
return ""
return VISION_TURN_HINT
def vision_debug_payload(result: VisionResult) -> dict[str, Any]:
from app.config import get_settings
payload: dict[str, Any] = {
"model": result.model,
"parsed": result.parsed,
"image_meta": result.image_meta,
"usage": result.usage,
"parse_error": result.parse_error,
}
if get_settings().vision_debug_enabled:
payload["raw_content"] = result.raw_content
return payload
def vision_debug_payloads(results: list[VisionResult]) -> dict[str, Any] | None:
if not results:
return None
items = [vision_debug_payload(result) for result in results]
if len(items) == 1:
return items[0]
models = {str(item.get("model") or "") for item in items}
payload: dict[str, Any] = {
"count": len(items),
"images": items,
"model": next(iter(models)) if len(models) == 1 else sorted(models),
}
return payload
backend/app/vision/preprocess.py
+53
View File
@@ -0,0 +1,53 @@
from __future__ import annotations
import io
from dataclasses import dataclass
from PIL import Image, ImageOps
from app.config import get_settings
@dataclass
class PreparedImage:
jpeg_bytes: bytes
width: int
height: int
original_bytes: int
compressed_bytes: int
mime: str = "image/jpeg"
def to_meta(self) -> dict[str, int | str]:
return {
"mime": self.mime,
"width": self.width,
"height": self.height,
"original_bytes": self.original_bytes,
"compressed_bytes": self.compressed_bytes,
}
def prepare_image(raw_bytes: bytes) -> PreparedImage:
settings = get_settings()
max_edge = max(256, int(settings.vision_max_edge_px))
quality = max(40, min(95, int(settings.vision_jpeg_quality)))
with Image.open(io.BytesIO(raw_bytes)) as img:
img = ImageOps.exif_transpose(img)
img = img.convert("RGB")
width, height = img.size
if max(width, height) > max_edge:
img.thumbnail((max_edge, max_edge), Image.Resampling.LANCZOS)
width, height = img.size
buffer = io.BytesIO()
img.save(buffer, format="JPEG", quality=quality, optimize=True)
jpeg_bytes = buffer.getvalue()
return PreparedImage(
jpeg_bytes=jpeg_bytes,
width=width,
height=height,
original_bytes=len(raw_bytes),
compressed_bytes=len(jpeg_bytes),
)
backend/app/vision/prompts.py
+30
View File
@@ -0,0 +1,30 @@
VISION_SYSTEM_PROMPT = """
Ты OCR-ассистент для скриншотов приложений здоровья и фитнеса (Mi Fitness, Xiaomi, Zepp Life и аналоги).
Извлеки ВСЕ видимые тексты, числа и таблицы. Приоритет — измеримые данные: длительность, калории, пульс, шаги, дистанция, дата, название активности.
Ответ — ТОЛЬКО JSON без markdown и комментариев.
Схема:
{
"description": "краткое описание экрана",
"document_type": "fitness_workout|fitness_steps|fitness_summary|other",
"extracted_text": ["строка1"],
"tables": [{"title": "заголовок или null", "rows": [["ячейка1", "ячейка2"]]}],
"fitness_hints": {
"title": null,
"activity_type": null,
"duration_min": null,
"active_calories": null,
"total_calories": null,
"steps": null,
"avg_heart_rate": null,
"date": null
},
"confidence": "high|medium|low",
"notes": ""
}
Правила:
- extracted_text — все значимые строки с экрана по порядку сверху вниз.
- tables — любые табличные блоки (заголовок + строки).
- fitness_hints — только если данные явно видны; иначе null.
- duration_min — целые минуты; steps — целое число; калории и пульс — числа.
- confidence=low если текст размыт или часть обрезана.
""".strip()
backend/app/vision/storage.py
+32
View File
@@ -0,0 +1,32 @@
from __future__ import annotations
import uuid
from pathlib import Path
from app.config import get_settings
from app.vision.preprocess import PreparedImage
def save_upload(prepared: PreparedImage, *, user_id: int) -> str:
settings = get_settings()
user_dir = Path(settings.uploads_dir) / str(user_id)
user_dir.mkdir(parents=True, exist_ok=True)
name = f"{uuid.uuid4().hex}.jpg"
path = user_dir / name
path.write_bytes(prepared.jpeg_bytes)
return name
def upload_media_path(user_id: int, filename: str) -> str:
return f"/api/v1/media/uploads/{user_id}/{filename}"
def format_upload_images_markdown(user_id: int, filenames: list[str]) -> str:
if not filenames:
return ""
total = len(filenames)
lines: list[str] = []
for index, name in enumerate(filenames, start=1):
alt = f"скриншот {index}/{total}" if total > 1 else "скриншот"
lines.append(f"![{alt}]({upload_media_path(user_id, name)})")
return "\n".join(lines)
backend/prompts/assistant.md
+7 -1
View File
@@ -22,7 +22,8 @@
- В ответах пользователю не используй эмодзи
Погода и дайджест:
- Вопросы о погоде, дожде, «что на улице» — используй get_weather или данные из блока [Погода]
- Вопросы о погоде, дожде, «что на улице», «завтра», «на неделю» — get_weather (hours_ahead, days_ahead)
- Блок [Погода] в контексте появляется только при релевантном запросе
- Утренний брифинг — get_morning_briefing
Списки покупок:
@@ -33,3 +34,8 @@
- «Нарисуй себя» → generate_image draw_self=true; «в полный рост» → scene_description="full_body, standing"
- Другая сцена → scene_description (booru-теги или короткий запрос); draw_self=true если персонаж из карточки
- Внешность персонажа задаётся в настройках карточки, не выдумывай теги
Скриншоты:
- Пользователь может прикрепить фото; vision-модель уже разобрала его до твоего ответа
- Результат — блок [Скриншот: ...] в сообщении: Описание, текст с экрана, fitness_hints
- Отвечай по этому блоку как по увиденному; не говори, что не видишь картинку, и не предлагай настроить Gemini/OpenRouter
backend/requirements.txt
+1
View File
@@ -8,4 +8,5 @@ python-dotenv>=1.0.1
aiosqlite>=0.20.0
httpx>=0.28.0
feedparser>=6.0.11
Pillow>=11.0.0
qdrant-client>=1.12.0,<1.13.0
backend/tests/test_activity_budget.py
-84
View File
@@ -1,84 +0,0 @@
import unittest
from app.fitness.activity_budget import (
compute_activity_bonus,
scale_targets,
steps_bonus_kcal,
)
PROFILE = {
"weight_kg": 70,
"activity_level": "moderate",
"weekly_workouts": 3,
"calorie_target": 2000,
"protein_g": 126,
"fat_g": 56,
"carbs_g": 250,
"water_l": 2.5,
}
class ActivityBudgetTests(unittest.TestCase):
def test_no_bonus_at_baseline(self) -> None:
bonus = compute_activity_bonus(
PROFILE,
steps_total=9000,
workouts=[{"active_calories": 85}],
)
self.assertEqual(bonus.steps_bonus_kcal, 0.0)
self.assertEqual(bonus.workout_bonus_kcal, 0.0)
self.assertEqual(bonus.total_bonus_kcal, 0.0)
self.assertEqual(bonus.scale_factor, 1.0)
def test_steps_and_workout_bonus(self) -> None:
bonus = compute_activity_bonus(
PROFILE,
steps_total=21800,
workouts=[{"active_calories": 155}],
)
self.assertGreater(bonus.steps_bonus_kcal, 0)
self.assertGreater(bonus.workout_bonus_kcal, 0)
self.assertEqual(
bonus.total_bonus_kcal,
round(bonus.steps_bonus_kcal + bonus.workout_bonus_kcal, 1),
)
self.assertGreater(bonus.scale_factor, 1.0)
def test_steps_bonus_formula(self) -> None:
kcal = steps_bonus_kcal(steps=21800, baseline_steps=9000, weight_kg=70)
self.assertEqual(kcal, round(12800 * 70 * 0.0005, 1))
def test_proportional_macro_scale(self) -> None:
base = {
"calories": 2000,
"protein_g": 100,
"fat_g": 50,
"carbs_g": 200,
"water_ml": 2500,
}
effective, targets_base = scale_targets(base, 500)
self.assertEqual(effective["calories"], 2500)
self.assertEqual(targets_base, base)
self.assertEqual(effective["water_ml"], 2500)
ratio = effective["calories"] / base["calories"]
self.assertAlmostEqual(effective["protein_g"] / base["protein_g"], ratio, places=1)
self.assertAlmostEqual(effective["fat_g"] / base["fat_g"], ratio, places=1)
self.assertAlmostEqual(effective["carbs_g"] / base["carbs_g"], ratio, places=1)
def test_floor_at_base_when_no_activity(self) -> None:
effective, _ = scale_targets(
{
"calories": 2045,
"protein_g": 156,
"fat_g": 57,
"carbs_g": 227,
"water_ml": 2900,
},
0,
)
self.assertEqual(effective["calories"], 2045)
if __name__ == "__main__":
unittest.main()
backend/tests/test_openmeteo_helpers.py
+58 -14
View File
@@ -1,13 +1,17 @@
from unittest.mock import patch
from app.homelab.openmeteo import (
PRECIP_PROB_HINT,
RECOMMENDED_SYNC_DOMAINS,
RECOMMENDED_SYNC_VARIABLES,
SYNC_HINT,
_coverage_sufficient,
_field_coverage,
_hourly_start_index,
_local_needs_sync_hint,
build_weather_dashboard,
format_weather_snapshot,
weather_query_relevant,
)
@@ -21,49 +25,89 @@ def test_coverage_sufficient():
assert _coverage_sufficient(
{
"current": ["temperature_2m", "weather_code", "wind_speed_10m"],
"hourly": ["temperature_2m", "precipitation_probability", "weather_code"],
"hourly": ["temperature_2m", "weather_code"],
}
) is True
def test_field_coverage_partial():
raw = {
"current": {"time": "2026-06-14T18:15", "temperature_2m": 20.6},
"current": {"time": "2026-06-14T18:15", "temperature_2m": 20.6, "weather_code": 2},
"hourly": {
"time": ["2026-06-14T18:00", "2026-06-14T19:00"],
"temperature_2m": [20.0, 19.5],
"precipitation": [0.0, 0.0],
"weather_code": [2, 3],
},
"daily": {
"time": ["2026-06-14", "2026-06-15"],
"temperature_2m_max": [21.0, 18.0],
"temperature_2m_min": [12.0, 10.0],
"weather_code": [2, 3],
},
}
coverage = _field_coverage(raw)
assert coverage["current"] == ["temperature_2m"]
assert "temperature_2m" in coverage["hourly"]
assert "precipitation" in coverage["hourly"]
assert "weather_code" not in coverage["hourly"]
assert "temperature_2m" in coverage["current"]
assert "weather_code" in coverage["hourly"]
assert "temperature_2m_max" in coverage["daily"]
def test_build_weather_dashboard_includes_sync_hint():
def test_local_needs_sync_hint():
assert _local_needs_sync_hint({"current": ["temperature_2m"], "hourly": ["temperature_2m"]}) is True
assert _local_needs_sync_hint(
{"current": ["temperature_2m", "weather_code"], "hourly": ["temperature_2m", "weather_code"]}
) is False
def test_weather_query_relevant():
assert weather_query_relevant("какая погода завтра")
assert not weather_query_relevant("напиши код на python")
def test_format_weather_snapshot_includes_tomorrow():
snap = {
"ok": True,
"location": "СПб",
"current": {"temperature_c": 20, "conditions": "ясно"},
"hourly": [],
"daily": [
{"label": "Сегодня", "temperature_min_c": 10, "temperature_max_c": 20, "conditions": "ясно"},
{"label": "Завтра", "temperature_min_c": 12, "temperature_max_c": 18, "conditions": "дождь"},
],
}
text = format_weather_snapshot(snap)
assert "Завтра:" in text
assert "None" not in text
def test_build_weather_dashboard_includes_daily():
fake_weather = {
"ok": True,
"location": "Test",
"data_source": "local",
"local_field_coverage": {"current": ["temperature_2m"], "hourly": ["temperature_2m"]},
"field_coverage": {"current": ["temperature_2m"], "hourly": ["temperature_2m"]},
"sync_hint": SYNC_HINT,
"current": {"temperature_c": 10, "conditions": "неизвестно"},
"local_field_coverage": {"current": ["temperature_2m", "weather_code"], "hourly": ["temperature_2m"], "daily": []},
"field_coverage": {"current": ["temperature_2m"], "hourly": ["temperature_2m"], "daily": []},
"sync_hint": PRECIP_PROB_HINT,
"merged_fields": [],
"current": {"temperature_c": 10, "conditions": "ясно"},
"hourly": [],
"daily": [{"label": "Завтра", "temperature_min_c": 5, "temperature_max_c": 12, "conditions": "дождь"}],
}
with patch("app.homelab.openmeteo.OpenMeteoClient") as mock_cls:
client = mock_cls.return_value
client.fetch_current_and_hourly.return_value = fake_weather
client.fetch_forecast.return_value = fake_weather
client.rain_summary.return_value = "ok"
client.daily_summary.return_value = "Завтра: 5–12°C"
client.cache_status.return_value = {"source": "local", "has_data": True, "cached": True, "ttl_sec": 300}
client.location_name = "Test"
client.lat = 1.0
client.lon = 2.0
client.base_url = "http://local"
client.cache_ttl = 300
result = build_weather_dashboard()
assert result["sync_hint"]
client.forecast_days = 7
result = build_weather_dashboard(days_ahead=7)
assert result["daily_summary"] == "Завтра: 5–12°C"
assert result["recommended_sync"]["domains"] == RECOMMENDED_SYNC_DOMAINS
assert result["recommended_sync"]["variables"] == RECOMMENDED_SYNC_VARIABLES
assert SYNC_HINT # constant exists
backend/tests/test_tdee_backfill.py
+116
View File
@@ -0,0 +1,116 @@
import unittest
from unittest.mock import patch
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from app.db.base import Base
from app.db.models import FitnessProfile, User
from app.fitness.calculators import compute_targets
class TdeeBackfillTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine)
with self.engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE IF NOT EXISTS _schema_migrations ("
"name TEXT PRIMARY KEY, "
"applied_at DATETIME DEFAULT CURRENT_TIMESTAMP)"
)
)
def _insert_legacy_profile(self) -> None:
Session = sessionmaker(bind=self.engine)
with Session() as db:
user = User(username="tester", api_token_hash="x")
db.add(user)
db.flush()
db.add(
FitnessProfile(
user_id=user.id,
sex="male",
age=30,
height_cm=180,
weight_kg=86,
goal="maintain",
activity_level="active",
calorie_target=3200,
protein_g=155,
fat_g=89,
carbs_g=400,
water_l=3.0,
neat_base_kcal=None,
)
)
db.commit()
def test_backfill_recalculates_stored_targets(self) -> None:
from app.db import migrate_fitness
self._insert_legacy_profile()
with patch.object(migrate_fitness, "engine", self.engine):
updated = migrate_fitness.backfill_tdee_targets(force=True)
self.assertEqual(updated, 1)
Session = sessionmaker(bind=self.engine)
with Session() as db:
row = db.query(FitnessProfile).one()
expected = compute_targets(
{
"sex": row.sex,
"age": row.age,
"height_cm": row.height_cm,
"weight_kg": row.weight_kg,
"goal": row.goal,
"neat_base_kcal": 200,
}
)
self.assertEqual(row.neat_base_kcal, 200.0)
self.assertEqual(row.calorie_target, expected["calorie_target"])
self.assertEqual(row.protein_g, expected["protein_g"])
self.assertLess(row.calorie_target, 3200)
def test_backfill_runs_once(self) -> None:
from app.db import migrate_fitness
self._insert_legacy_profile()
with patch.object(migrate_fitness, "engine", self.engine):
first = migrate_fitness.backfill_tdee_targets(force=True)
second = migrate_fitness.backfill_tdee_targets()
self.assertEqual(first, 1)
self.assertEqual(second, 0)
def test_macros_backfill_updates_bju_only(self) -> None:
from app.db import migrate_fitness
from app.fitness.calculators import macro_targets
self._insert_legacy_profile()
with patch.object(migrate_fitness, "engine", self.engine):
migrate_fitness.backfill_tdee_targets(force=True)
Session = sessionmaker(bind=self.engine)
with Session() as db:
row = db.query(FitnessProfile).one()
calorie_before = row.calorie_target
row.protein_g = 999
db.commit()
updated = migrate_fitness.backfill_macros_gkg(force=True)
self.assertEqual(updated, 1)
with Session() as db:
row = db.query(FitnessProfile).one()
expected = macro_targets(calorie_before, row.weight_kg, row.goal)
self.assertEqual(row.calorie_target, calorie_before)
self.assertEqual(row.protein_g, expected["protein_g"])
self.assertEqual(row.fat_g, expected["fat_g"])
self.assertEqual(row.carbs_g, expected["carbs_g"])
if __name__ == "__main__":
unittest.main()
backend/tests/test_tdee_components.py
+126
View File
@@ -0,0 +1,126 @@
import unittest
from app.fitness.activity_budget import estimate_workout_active_kcal, infer_met, workouts_kcal_total
from app.fitness.calculators import (
DEFAULT_NEAT_KCAL,
bmr_mifflin,
compute_daily_targets,
compute_targets,
compute_tdee,
macro_targets,
steps_kcal,
water_target_l,
)
PROFILE = {
"sex": "male",
"age": 30,
"height_cm": 180,
"weight_kg": 86,
"goal": "maintain",
"neat_base_kcal": 200,
}
class TdeeComponentsTests(unittest.TestCase):
def test_rest_day_tdee_is_bmr_plus_neat(self) -> None:
breakdown = compute_tdee(PROFILE, steps_total=0, workouts=[])
bmr = bmr_mifflin(sex="male", weight_kg=86, height_cm=180, age=30)
self.assertEqual(breakdown["bmr"], round(bmr, 0))
self.assertEqual(breakdown["neat_kcal"], DEFAULT_NEAT_KCAL)
self.assertEqual(breakdown["steps_kcal"], 0.0)
self.assertEqual(breakdown["workout_kcal"], 0.0)
self.assertEqual(breakdown["tdee"], round(bmr + DEFAULT_NEAT_KCAL, 0))
def test_steps_kcal_at_reference_weight(self) -> None:
kcal = steps_kcal(steps=10000, weight_kg=86)
self.assertAlmostEqual(kcal, 400.0, delta=1.0)
def test_daily_targets_include_activity(self) -> None:
daily = compute_daily_targets(
PROFILE,
steps_total=8000,
workouts=[{"active_calories": 450}],
)
self.assertGreater(daily["steps_kcal"], 0)
self.assertEqual(daily["workout_kcal"], 450.0)
self.assertEqual(
daily["tdee"],
daily["bmr"] + daily["neat_kcal"] + daily["steps_kcal"] + daily["workout_kcal"],
)
self.assertEqual(daily["calorie_target"], daily["tdee"])
def test_compute_targets_rest_day(self) -> None:
targets = compute_targets(PROFILE)
self.assertEqual(targets["steps_kcal"], 0)
self.assertEqual(targets["workout_kcal"], 0)
self.assertEqual(targets["calorie_target"], targets["tdee"])
def test_water_target(self) -> None:
self.assertEqual(water_target_l(70), 2.3)
def test_workout_active_calories_priority(self) -> None:
kcal = estimate_workout_active_kcal(
{"active_calories": 300, "duration_min": 60, "met": 9.8},
weight_kg=86,
)
self.assertEqual(kcal, 300.0)
def test_workout_met_fallback(self) -> None:
kcal = estimate_workout_active_kcal(
{"title": "бег", "duration_min": 60},
weight_kg=86,
)
self.assertAlmostEqual(kcal, 9.8 * 86, delta=1.0)
def test_workout_no_data_returns_zero(self) -> None:
self.assertEqual(estimate_workout_active_kcal({}, weight_kg=70), 0.0)
def test_infer_met_from_title(self) -> None:
self.assertEqual(infer_met({"title": "пробежал триатлон"}), 10.0)
def test_workouts_kcal_total(self) -> None:
total = workouts_kcal_total(
[
{"active_calories": 100},
{"title": "ходьба", "duration_min": 30},
],
weight_kg=86,
)
self.assertGreater(total, 100)
class MacroTargetsTests(unittest.TestCase):
def test_lose_macros_from_weight(self) -> None:
macros = macro_targets(2363, 86, "lose")
self.assertEqual(macros["protein_g"], 189)
self.assertEqual(macros["fat_g"], 86)
self.assertEqual(macros["carbs_g"], 208)
def test_maintain_macros_from_weight(self) -> None:
macros = macro_targets(2000, 86, "maintain")
self.assertEqual(macros["protein_g"], 155)
self.assertEqual(macros["fat_g"], 86)
self.assertEqual(macros["carbs_g"], 152)
def test_active_day_increases_carbs_only(self) -> None:
rest = compute_daily_targets(PROFILE, steps_total=0, workouts=[])
active = compute_daily_targets(
PROFILE,
steps_total=8000,
workouts=[{"active_calories": 450}],
)
self.assertEqual(rest["protein_g"], active["protein_g"])
self.assertEqual(rest["fat_g"], active["fat_g"])
self.assertGreater(active["calorie_target"], rest["calorie_target"])
self.assertGreater(active["carbs_g"], rest["carbs_g"])
def test_low_calorie_target_floors_carbs_at_zero(self) -> None:
macros = macro_targets(1000, 86, "lose")
self.assertEqual(macros["protein_g"], 189)
self.assertEqual(macros["fat_g"], 86)
self.assertEqual(macros["carbs_g"], 0)
if __name__ == "__main__":
unittest.main()
backend/tests/test_vision_analyze.py
+71
View File
@@ -0,0 +1,71 @@
import json
import unittest
from unittest.mock import AsyncMock, patch
from app.vision.analyze import VisionService, format_user_message, format_vision_turn_hint
from app.vision.preprocess import PreparedImage
class VisionAnalyzeTests(unittest.TestCase):
def test_format_user_message_with_fitness_hints(self) -> None:
from app.vision.analyze import VisionResult
result = VisionResult(
parsed={
"description": "Экран тренировки бег",
"document_type": "fitness_workout",
"extracted_text": ["45 мин", "420 ккал"],
"tables": [{"title": "Пульс", "rows": [["средний", "152"]]}],
"fitness_hints": {"duration_min": 45, "active_calories": 420},
"confidence": "high",
},
raw_content="{}",
model="test-model",
)
text = format_user_message("запиши тренировку", result)
self.assertIn("[Скриншот: fitness_workout", text)
self.assertIn("420 ккал", text)
self.assertIn("Подпись: запиши тренировку", text)
def test_run_async_analyze(self) -> None:
import asyncio
prepared = PreparedImage(
jpeg_bytes=b"fakejpeg",
width=100,
height=100,
original_bytes=200,
compressed_bytes=100,
)
payload = {
"description": "Шаги за день",
"document_type": "fitness_steps",
"extracted_text": ["8432 шага"],
"tables": [],
"fitness_hints": {"steps": 8432},
"confidence": "high",
"notes": "",
}
async def _run() -> None:
service = VisionService()
with patch.object(
service.llm,
"complete_vision",
new=AsyncMock(return_value={"content": json.dumps(payload), "model": "vision-test", "usage": {}}),
):
result = await service.analyze_prepared(prepared, user_hint="шаги")
self.assertEqual(result.parsed["document_type"], "fitness_steps")
self.assertEqual(result.parsed["fitness_hints"]["steps"], 8432)
self.assertEqual(result.model, "vision-test")
asyncio.run(_run())
def test_format_vision_turn_hint(self) -> None:
self.assertEqual(format_vision_turn_hint("привет"), "")
self.assertIn("не видишь", format_vision_turn_hint("[Скриншот: other, confidence=high]\nОписание: test"))
self.assertIn("не видишь", format_vision_turn_hint("[Скриншот 2/3: other, confidence=high]\nОписание: test"))
if __name__ == "__main__":
unittest.main()
backend/tests/test_vision_format.py
+41
View File
@@ -0,0 +1,41 @@
import unittest
from app.vision.analyze import VisionResult, format_user_message, format_user_messages, vision_debug_payloads
class FormatUserMessageTests(unittest.TestCase):
def test_parse_error_in_message(self) -> None:
result = VisionResult(parsed={}, raw_content="not json", parse_error="bad json")
text = format_user_message("", result)
self.assertIn("parse_error", text)
def test_multiple_screenshots_numbered(self) -> None:
results = [
VisionResult(
parsed={"document_type": "fitness_steps", "confidence": "high", "description": "Шаги"},
raw_content="{}",
),
VisionResult(
parsed={"document_type": "other", "confidence": "medium", "description": "Еда"},
raw_content="{}",
),
]
text = format_user_messages("сравни", results)
self.assertIn("[Скриншот 1/2: fitness_steps", text)
self.assertIn("[Скриншот 2/2: other", text)
self.assertIn("Подпись: сравни", text)
self.assertEqual(text.count("Подпись:"), 1)
def test_vision_debug_payloads_multi(self) -> None:
results = [
VisionResult(parsed={"document_type": "other"}, raw_content="{}", model="m1"),
VisionResult(parsed={"document_type": "other"}, raw_content="{}", model="m1"),
]
payload = vision_debug_payloads(results)
assert payload is not None
self.assertEqual(payload["count"], 2)
self.assertEqual(len(payload["images"]), 2)
if __name__ == "__main__":
unittest.main()
backend/tests/test_vision_preprocess.py
+30
View File
@@ -0,0 +1,30 @@
import io
import unittest
from PIL import Image
from app.vision.preprocess import prepare_image
class VisionPreprocessTests(unittest.TestCase):
def _make_png(self, width: int, height: int) -> bytes:
buffer = io.BytesIO()
Image.new("RGB", (width, height), color=(120, 80, 200)).save(buffer, format="PNG")
return buffer.getvalue()
def test_resize_large_image(self) -> None:
raw = self._make_png(2400, 1600)
prepared = prepare_image(raw)
self.assertLessEqual(max(prepared.width, prepared.height), 1280)
self.assertLess(prepared.compressed_bytes, prepared.original_bytes)
def test_small_image_keeps_dimensions(self) -> None:
raw = self._make_png(640, 480)
prepared = prepare_image(raw)
self.assertEqual(prepared.width, 640)
self.assertEqual(prepared.height, 480)
self.assertEqual(prepared.mime, "image/jpeg")
if __name__ == "__main__":
unittest.main()
backend/tests/test_vision_storage.py
+19
View File
@@ -0,0 +1,19 @@
import unittest
from app.vision.storage import format_upload_images_markdown, upload_media_path
class UploadMarkdownTests(unittest.TestCase):
def test_single_image(self) -> None:
md = format_upload_images_markdown(5, ["abc.jpg"])
self.assertIn(upload_media_path(5, "abc.jpg"), md)
self.assertIn("![скриншот]", md)
def test_multiple_images(self) -> None:
md = format_upload_images_markdown(3, ["a.jpg", "b.jpg"])
self.assertIn("скриншот 1/2", md)
self.assertIn("скриншот 2/2", md)
if __name__ == "__main__":
unittest.main()
backend/tests/test_weather_dashboard.py
+22 -4
View File
@@ -7,6 +7,11 @@ def test_build_weather_dashboard_structure():
fake_weather = {
"ok": True,
"location": "Test City",
"data_source": "local",
"local_field_coverage": {"current": ["temperature_2m"], "hourly": [], "daily": []},
"field_coverage": {"current": ["temperature_2m"], "hourly": [], "daily": []},
"sync_hint": "",
"merged_fields": [],
"current": {
"time": "2026-06-13T12:00",
"temperature_c": 18.5,
@@ -27,12 +32,22 @@ def test_build_weather_dashboard_structure():
"conditions": "переменная облачность",
}
],
"daily": [
{
"date": "2026-06-14",
"label": "Завтра",
"temperature_max_c": 20.0,
"temperature_min_c": 12.0,
"conditions": "дождь",
}
],
}
with patch("app.homelab.openmeteo.OpenMeteoClient") as mock_cls:
client = mock_cls.return_value
client.fetch_current_and_hourly.return_value = fake_weather
client.fetch_forecast.return_value = fake_weather
client.rain_summary.return_value = "Существенных осадков в ближайшие часы не ожидается."
client.daily_summary.return_value = "Завтра: 12–20°C"
client.cache_status.return_value = {
"has_data": True,
"cached": True,
@@ -40,18 +55,21 @@ def test_build_weather_dashboard_structure():
"age_sec": 10,
"ttl_sec": 300,
"expires_in_sec": 290,
"source": "local",
"merged_fields": [],
}
client.location_name = "Test City"
client.lat = 59.9
client.lon = 30.3
client.base_url = "http://openmeteo.test"
client.cache_ttl = 300
client.forecast_days = 7
result = build_weather_dashboard(hours_ahead=6)
result = build_weather_dashboard(hours_ahead=6, days_ahead=7)
assert result["weather"]["ok"] is True
assert "[Погода]" in result["assistant_context"]
assert "None" not in result["assistant_context"]
assert "temperature_2m" in result["available_fields"]["current"]
assert "get_weather" in result["assistant_tools"]
assert "daily" in result["available_fields"]
assert result["daily_summary"]
assert result["config"]["location"] == "Test City"