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    sys.path.extend(str(x) for x in search_path)
158
159    os.environ["CBRKIT_RETRIEVER"] = ",".join(retriever)
160    os.environ["CBRKIT_REUSER"] = ",".join(reuser)
161    os.environ["CBRKIT_SYNTHESIZER"] = ",".join(synthesizer)
162
163    uvicorn.run(
164        "cbrkit.api:app",
165        host=host,
166        port=port,
167        reload=reload,
168        root_path=root_path,
169    )
170
171
172@app.command()
173def openapi(file: Path | None = None):
174    from cbrkit.api import app
175
176    schema = orjson.dumps(
177        app.openapi(),
178        option=orjson.OPT_INDENT_2,
179    )
180
181    if file is None:
182        print(schema.decode())
183
184    else:
185        print(f"Writing OpenAPI schema to {file}")
186
187        with file.open("wb") as fp:
188            fp.write(schema)
189
190
191if __name__ == "__main__":
192    app()