1"""
  2.. include:: ../../cli.md
  3"""
  4
  5import os
  6import sys
  7from pathlib import Path
  8from typing import Annotated, Any
  9
 10import orjson
 11
 12import cbrkit
 13
 14with cbrkit.helpers.optional_dependencies("raise", "cli"):
 15    import typer
 16    from rich import print
 17
 18
 19__all__ = ["app"]
 20
 21app = typer.Typer(pretty_exceptions_enable=False)
 22
 23
 24@app.callback()
 25def app_callback():
 26    """Initialize the CBRKit CLI application."""
 27    pass
 28
 29
 30@app.command()
 31def retrieve(
 32    casebase_path: Path,
 33    queries_path: Path,
 34    retriever: str,
 35    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 36    print_ranking: bool = True,
 37    print_similarities: bool = False,
 38    output_path: Path | None = None,
 39) -> None:
 40    """Retrieve similar cases from a casebase for the given queries."""
 41    sys.path.extend(str(x) for x in search_path)
 42    casebase = cbrkit.loaders.path(casebase_path)
 43    queries = cbrkit.loaders.path(queries_path)
 44    retrievers: list[
 45        cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc[Any, Any, Any]]
 46    ] = cbrkit.helpers.load_callables(retriever)
 47
 48    result = cbrkit.retrieval.apply_queries(casebase, queries, retrievers)
 49
 50    if output_path:
 51        cbrkit.dumpers.file(output_path, result)
 52
 53    if print_ranking or print_similarities:
 54        for query_key, query_result in result.final_step.queries.items():
 55            print(f"Query: {query_key}")
 56
 57            if print_ranking:
 58                print(f"Ranking: {', '.join(map(str, query_result.ranking))}")
 59
 60            if print_similarities:
 61                print("Similarities:")
 62                for case_name, similarity in query_result.similarities.items():
 63                    print(f"  {case_name}: {cbrkit.helpers.unpack_float(similarity)}")
 64
 65            print()
 66
 67
 68@app.command()
 69def reuse(
 70    casebase_path: Path,
 71    queries_path: Path,
 72    reuser: str,
 73    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 74    output_path: Path | None = None,
 75) -> None:
 76    """Reuse retrieved cases by adapting them for the given queries."""
 77    sys.path.extend(str(x) for x in search_path)
 78    casebase = cbrkit.loaders.path(casebase_path)
 79    queries = cbrkit.loaders.path(queries_path)
 80    reusers: list[
 81        cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc[Any, Any, Any]]
 82    ] = cbrkit.helpers.load_callables(reuser)
 83
 84    result = cbrkit.reuse.apply_queries(casebase, queries, reusers)
 85
 86    if output_path:
 87        cbrkit.dumpers.file(output_path, result)
 88
 89
 90@app.command()
 91def cycle(
 92    casebase_path: Path,
 93    queries_path: Path,
 94    retriever: str,
 95    reuser: str,
 96    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 97    reviser: str | None = None,
 98    retainer: str | None = None,
 99    output_path: Path | None = None,
100) -> None:
101    """Run a full CBR cycle with retrieval, reuse, revision, and retention."""
102    sys.path.extend(str(x) for x in search_path)
103    casebase = cbrkit.loaders.path(casebase_path)
104    queries = cbrkit.loaders.path(queries_path)
105    retrievers: list[
106        cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc[Any, Any, Any]]
107    ] = cbrkit.helpers.load_callables(retriever)
108    reusers: list[
109        cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc[Any, Any, Any]]
110    ] = cbrkit.helpers.load_callables(reuser)
111    revisers: list[
112        cbrkit.typing.MaybeFactory[cbrkit.typing.ReviserFunc[Any, Any, Any]]
113    ] = cbrkit.helpers.load_callables(reviser) if reviser else []
114    retainers: list[
115        cbrkit.typing.MaybeFactory[cbrkit.typing.RetainerFunc[Any, Any, Any]]
116    ] = cbrkit.helpers.load_callables(retainer) if retainer else []
117
118    result = cbrkit.cycle.apply_queries(
119        casebase, queries, retrievers, reusers, revisers, retainers
120    )
121
122    if output_path:
123        cbrkit.dumpers.file(output_path, result)
124
125
126@app.command()
127def synthesis(
128    casebase_path: Path,
129    queries_path: Path,
130    retriever: str,
131    reuser: str,
132    synthesizer: str,
133    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
134    reviser: str | None = None,
135    retainer: str | None = None,
136    output_path: Path | None = None,
137) -> None:
138    """Run a CBR cycle followed by synthesis to produce a generated response."""
139    sys.path.extend(str(x) for x in search_path)
140    casebase = cbrkit.loaders.path(casebase_path)
141    queries = cbrkit.loaders.path(queries_path)
142    retrievers: list[
143        cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc[Any, Any, Any]]
144    ] = cbrkit.helpers.load_callables(retriever)
145    reusers: list[
146        cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc[Any, Any, Any]]
147    ] = cbrkit.helpers.load_callables(reuser)
148    revisers: list[
149        cbrkit.typing.MaybeFactory[cbrkit.typing.ReviserFunc[Any, Any, Any]]
150    ] = cbrkit.helpers.load_callables(reviser) if reviser else []
151    retainers: list[
152        cbrkit.typing.MaybeFactory[cbrkit.typing.RetainerFunc[Any, Any, Any]]
153    ] = cbrkit.helpers.load_callables(retainer) if retainer else []
154    synthesis_func: cbrkit.typing.MaybeFactory[
155        cbrkit.typing.SynthesizerFunc[Any, Any, Any, Any]
156    ] = cbrkit.helpers.load_callable(synthesizer)
157
158    cycle_result = cbrkit.cycle.apply_queries(
159        casebase, queries, retrievers, reusers, revisers, retainers
160    )
161    synthesis_result = cbrkit.synthesis.apply_result(
162        cycle_result.retain, synthesis_func
163    )
164
165    if output_path:
166        cbrkit.dumpers.file(output_path, synthesis_result)
167
168
169@app.command()
170def serve(
171    retriever: Annotated[list[str], typer.Option(default_factory=list)],
172    reuser: Annotated[list[str], typer.Option(default_factory=list)],
173    reviser: Annotated[list[str], typer.Option(default_factory=list)],
174    retainer: Annotated[list[str], typer.Option(default_factory=list)],
175    synthesizer: Annotated[list[str], typer.Option(default_factory=list)],
176    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
177    host: str = "0.0.0.0",
178    port: int = 8080,
179    reload: bool = False,
180    root_path: str = "",
181) -> None:
182    """Start the CBRKit API server with the specified components."""
183    import uvicorn
184
185    from cbrkit.api import app
186
187    sys.path.extend(str(x) for x in search_path)
188
189    os.environ["CBRKIT_RETRIEVER"] = ",".join(retriever)
190    os.environ["CBRKIT_REUSER"] = ",".join(reuser)
191    os.environ["CBRKIT_REVISER"] = ",".join(reviser)
192    os.environ["CBRKIT_RETAINER"] = ",".join(retainer)
193    os.environ["CBRKIT_SYNTHESIZER"] = ",".join(synthesizer)
194
195    uvicorn.run(
196        app,
197        host=host,
198        port=port,
199        reload=reload,
200        root_path=root_path,
201    )
202
203
204@app.command()
205def uvicorn(
206    app: str,
207    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
208    host: str = "0.0.0.0",
209    port: int = 8080,
210    reload: bool = False,
211    root_path: str = "",
212) -> None:
213    """Start a custom ASGI application using the uvicorn server."""
214    import uvicorn
215
216    sys.path.extend(str(x) for x in search_path)
217
218    uvicorn.run(
219        app,
220        host=host,
221        port=port,
222        reload=reload,
223        root_path=root_path,
224    )
225
226
227@app.command()
228def openapi(file: Path | None = None):
229    """Generate the OpenAPI schema and print or write it to a file."""
230    from cbrkit.api import app
231
232    schema = orjson.dumps(
233        app.openapi(),
234        option=orjson.OPT_INDENT_2,
235    )
236
237    if file is None:
238        print(schema.decode())
239
240    else:
241        print(f"Writing OpenAPI schema to {file}")
242
243        with file.open("wb") as fp:
244            fp.write(schema)
245
246
247if __name__ == "__main__":
248    app()