86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
"""Client typé du provider Kazeia, par tablette.
|
|
|
|
Enveloppe `Adb.content_query` et expose des accès typés (pydantic) aux endpoints
|
|
EXISTANTS du provider. Les écritures et la lecture du texte riche (config/prompts,
|
|
texte des conversations) dépendent des méthodes call() base64-JSON décrites dans
|
|
docs/PROVIDER_RPC_SPEC.md — non livrées côté app patiente → exposées ici comme
|
|
NotImplemented explicites pour cadrer l'intégration future.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from ..adb import Adb
|
|
from . import models as m
|
|
|
|
|
|
class ProviderClient:
|
|
def __init__(self, adb: Adb, serial: str) -> None:
|
|
self.adb = adb
|
|
self.serial = serial
|
|
|
|
def _q(self, path: str, **kw) -> list[dict]:
|
|
return self.adb.content_query(path, serial=self.serial, **kw)
|
|
|
|
# ---- lecture (endpoints actuels, valeurs simples) ----------------------
|
|
def state(self) -> m.StateRow:
|
|
return m.StateRow(**self._q("state")[0])
|
|
|
|
def turns(self) -> list[m.TurnRow]:
|
|
return [m.TurnRow(**r) for r in self._q("turns")]
|
|
|
|
def crashes(self) -> list[m.CrashRow]:
|
|
return [m.CrashRow(**r) for r in self._q("crashes")]
|
|
|
|
def models(self) -> list[m.ModelRow]:
|
|
return [m.ModelRow(**r) for r in self._q("models")]
|
|
|
|
def voices(self) -> list[m.VoiceRow]:
|
|
return [m.VoiceRow(**r) for r in self._q("voices")]
|
|
|
|
def rag_status(self) -> m.RagStatus:
|
|
return m.RagStatus(**self._q("rag_status")[0])
|
|
|
|
def rag_index(self) -> list[m.RagDocRow]:
|
|
return [m.RagDocRow(**r) for r in self._q("rag")]
|
|
|
|
def rag_query(self, text: str) -> list[m.RagHit]:
|
|
# text peut contenir des espaces → fragile via query ; OK pour un test simple.
|
|
return [m.RagHit(**r) for r in self._q("rag_query", where=text)]
|
|
|
|
def updates(self) -> m.UpdateStatus:
|
|
return m.UpdateStatus(**self._q("updates")[0])
|
|
|
|
def dist_config(self) -> m.DistConfig:
|
|
return m.DistConfig(**self._q("dist_config")[0])
|
|
|
|
def sessions(self, profile_id: str | None = None) -> list[m.SessionRow]:
|
|
path = f"conversations/profile/{profile_id}" if profile_id else "conversations"
|
|
return [m.SessionRow(**r) for r in self._q(path)]
|
|
|
|
def profiles(self) -> list[m.ProfileRow]:
|
|
return [m.ProfileRow(**r) for r in self._q("profiles")]
|
|
|
|
def active_profile_id(self) -> str | None:
|
|
rows = self._q("profiles/active")
|
|
return rows[0].get("active_profile_id") if rows else None
|
|
|
|
# ---- MAJ OTA (call() déjà disponibles côté app) ------------------------
|
|
def update_check(self) -> dict:
|
|
return self.adb.content_call("update_check", serial=self.serial)
|
|
|
|
def update_install(self) -> dict:
|
|
return self.adb.content_call("update_install", serial=self.serial)
|
|
|
|
# ---- bloqué tant que PROVIDER_RPC_SPEC n'est pas livré côté app ---------
|
|
def dump_config(self):
|
|
raise NotImplementedError(
|
|
"Nécessite la méthode call() `cfg_dump_json` (PROVIDER_RPC_SPEC §2.1) "
|
|
"— non livrée côté app patiente."
|
|
)
|
|
|
|
def session_turns(self, session_id: str):
|
|
raise NotImplementedError(
|
|
"Le texte des tours contient des sauts de ligne → `content query` casse. "
|
|
"Passer par `conversations_export` (PROVIDER_RPC_SPEC §4) une fois livré."
|
|
)
|