1"""
  2.. include:: ../../cli.md
  3"""
  4
  5import os
  6import sys
  7from pathlib import Path
  8from typing import Annotated
  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    pass
 27
 28
 29@app.command()
 30def retrieve(
 31    casebase_path: Path,
 32    queries_path: Path,
 33    retriever: str,
 34    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 35    print_ranking: bool = True,
 36    print_similarities: bool = False,
 37    output_path: Path | None = None,
 38) -> None:
 39    sys.path.extend(str(x) for x in search_path)
 40    casebase = cbrkit.loaders.path(casebase_path)
 41    queries = cbrkit.loaders.path(queries_path)
 42    retrievers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc]] = (
 43        cbrkit.helpers.load_callables(retriever)
 44    )
 45
 46    result = cbrkit.retrieval.apply_queries(casebase, queries, retrievers)
 47
 48    if output_path:
 49        cbrkit.dumpers.file(output_path, result)
 50
 51    if print_ranking or print_similarities:
 52        for query_key, query_result in result.final_step.queries.items():
 53            print(f"Query: {query_key}")
 54
 55            if print_ranking:
 56                print(f"Ranking: {', '.join(map(str, query_result.ranking))}")
 57
 58            if print_similarities:
 59                print("Similarities:")
 60                for case_name, similarity in query_result.similarities.items():
 61                    print(f"  {case_name}: {cbrkit.helpers.unpack_float(similarity)}")
 62
 63            print()
 64
 65
 66@app.command()
 67def reuse(
 68    casebase_path: Path,
 69    queries_path: Path,
 70    reuser: str,
 71    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 72    output_path: Path | None = None,
 73) -> None:
 74    sys.path.extend(str(x) for x in search_path)
 75    casebase = cbrkit.loaders.path(casebase_path)
 76    queries = cbrkit.loaders.path(queries_path)
 77    reusers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc]] = (
 78        cbrkit.helpers.load_callables(reuser)
 79    )
 80
 81    result = cbrkit.reuse.apply_queries(casebase, queries, reusers)
 82
 83    if output_path:
 84        cbrkit.dumpers.file(output_path, result)
 85
 86
 87@app.command()
 88def cycle(
 89    casebase_path: Path,
 90    queries_path: Path,
 91    retriever: str,
 92    reuser: str,
 93    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
 94    output_path: Path | None = None,
 95) -> None:
 96    sys.path.extend(str(x) for x in search_path)
 97    casebase = cbrkit.loaders.path(casebase_path)
 98    queries = cbrkit.loaders.path(queries_path)
 99    retrievers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc]] = (
100        cbrkit.helpers.load_callables(retriever)
101    )
102    reusers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc]] = (
103        cbrkit.helpers.load_callables(reuser)
104    )
105
106    result = cbrkit.cycle.apply_queries(casebase, queries, retrievers, reusers)
107
108    if output_path:
109        cbrkit.dumpers.file(output_path, result)
110
111
112@app.command()
113def synthesis(
114    casebase_path: Path,
115    queries_path: Path,
116    retriever: str,
117    reuser: str,
118    synthesizer: str,
119    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
120    output_path: Path | None = None,
121) -> None:
122    sys.path.extend(str(x) for x in search_path)
123    casebase = cbrkit.loaders.path(casebase_path)
124    queries = cbrkit.loaders.path(queries_path)
125    retrievers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.RetrieverFunc]] = (
126        cbrkit.helpers.load_callables(retriever)
127    )
128    reusers: list[cbrkit.typing.MaybeFactory[cbrkit.typing.ReuserFunc]] = (
129        cbrkit.helpers.load_callables(reuser)
130    )
131    synthesis_func: cbrkit.typing.MaybeFactory[cbrkit.typing.SynthesizerFunc] = (
132        cbrkit.helpers.load_callable(synthesizer)
133    )
134
135    cycle_result = cbrkit.cycle.apply_queries(casebase, queries, retrievers, reusers)
136    synthesis_result = cbrkit.synthesis.apply_result(
137        cycle_result.final_step, synthesis_func
138    )
139
140    if output_path:
141        cbrkit.dumpers.file(output_path, synthesis_result)
142
143
144@app.command()
145def serve(
146    retriever: Annotated[list[str], typer.Option(default_factory=list)],
147    reuser: Annotated[list[str], typer.Option(default_factory=list)],
148    synthesizer: Annotated[list[str], typer.Option(default_factory=list)],
149    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
150    host: str = "0.0.0.0",
151    port: int = 8080,
152    reload: bool = False,
153    root_path: str = "",
154) -> None:
155    import uvicorn
156
157    from cbrkit.api import app
158
159    sys.path.extend(str(x) for x in search_path)
160
161    os.environ["CBRKIT_RETRIEVER"] = ",".join(retriever)
162    os.environ["CBRKIT_REUSER"] = ",".join(reuser)
163    os.environ["CBRKIT_SYNTHESIZER"] = ",".join(synthesizer)
164
165    uvicorn.run(
166        app,
167        host=host,
168        port=port,
169        reload=reload,
170        root_path=root_path,
171    )
172
173
174@app.command()
175def uvicorn(
176    app: str,
177    search_path: Annotated[list[Path], typer.Option(default_factory=list)],
178    host: str = "0.0.0.0",
179    port: int = 8080,
180    reload: bool = False,
181    root_path: str = "",
182) -> None:
183    import uvicorn
184
185    sys.path.extend(str(x) for x in search_path)
186
187    uvicorn.run(
188        app,
189        host=host,
190        port=port,
191        reload=reload,
192        root_path=root_path,
193    )
194
195
196@app.command()
197def openapi(file: Path | None = None):
198    from cbrkit.api import app
199
200    schema = orjson.dumps(
201        app.openapi(),
202        option=orjson.OPT_INDENT_2,
203    )
204
205    if file is None:
206        print(schema.decode())
207
208    else:
209        print(f"Writing OpenAPI schema to {file}")
210
211        with file.open("wb") as fp:
212            fp.write(schema)
213
214
215if __name__ == "__main__":
216    app()