Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions backend/packages/acidwatch/src/acidwatch/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from __future__ import annotations

from acidwatch.client import Client

__all__ = ["Client"]
4 changes: 4 additions & 0 deletions backend/packages/acidwatch/src/acidwatch/__main__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
if __name__ == "__main__":
from acidwatch.cli import app

app()
51 changes: 0 additions & 51 deletions backend/packages/acidwatch/src/acidwatch/cli.py

This file was deleted.

17 changes: 17 additions & 0 deletions backend/packages/acidwatch/src/acidwatch/cli/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
from __future__ import annotations
from typer import Typer
from acidwatch.cli.model import model_app


app = Typer(
name="AcidWatch CLI",
no_args_is_help=True,
pretty_exceptions_enable=False,
rich_markup_mode=None,
)


app.add_typer(model_app)


__all__ = ["app"]
56 changes: 56 additions & 0 deletions backend/packages/acidwatch/src/acidwatch/cli/model.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
from __future__ import annotations
from rich.panel import Panel
from rich import print as rprint
from rich.table import Table
from rich.pretty import pretty_repr
from acidwatch import Client
from typer import Typer


model_app = Typer(name="model")


@model_app.command("list")
def model_list() -> None:
table = Table("ID", "Name", title="Available AcidWatch models")

with Client() as session:
for model in session.list_models():
if model.access_error is not None:
continue

table.add_row(model.model_id, model.display_name)

rprint(table)


@model_app.command("show")
def model_show(model_id: str) -> None:
with Client() as session:
for model in session.list_models():
if model.model_id == model_id:
break

else:
rprint("No such model: ", model_id)

info_table = Table(show_header=False)
info_table.add_row("Model ID", model.model_id)
info_table.add_row("Display name", model.display_name)
info_table.add_row("Valid substances", pretty_repr(model.valid_substances))
rprint(info_table)

if model.parameters:
parameter_table = Table(
"ID", "Label", "Type", "Default value", title="Parameters"
)
for param_id, param in model.parameters.items():
parameter_table.add_row(
param_id, param.label, param.type, str(param.default)
)
rprint(parameter_table)

rprint(Panel(model.description, title="Description", expand=False))


__all__ = ["model_app"]
62 changes: 62 additions & 0 deletions backend/packages/acidwatch/src/acidwatch/client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
from __future__ import annotations
from pydantic.alias_generators import to_camel
from pydantic import BaseModel, ConfigDict, RootModel, Field
from typing_extensions import Doc
from typing import Annotated, Any
import httpx


DEFAULT_API_URL = "https://backend-acidwatch-prod.radix.equinor.com"


class _Parameter(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
populate_by_name=True,
)

default: Any
label: str
unit: Annotated[str | None, Field(None)]
type: Annotated[str | None, Field(None)]


class Model(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
populate_by_name=True,
)

access_error: str | None
model_id: str
display_name: str
description: str

valid_substances: list[str]
parameters: dict[str, _Parameter]


class Client(httpx.Client):
def __init__(
self,
api_url: Annotated[
str,
Doc(
"""
AcidWatch API URL, pointing to the root of the "backend".
For local AcidWatch instance, use eg. "http://localhost:8000"
"""
),
] = DEFAULT_API_URL,
) -> None:
super().__init__(base_url=api_url)

def list_models(self) -> list[Model]:
resp = self.get("/models")

assert resp.status_code == 200

root_model = RootModel[list[Model]]
object = root_model.model_validate_json(resp.content)

return object.root
Loading