#!/usr/bin/env python from collections import defaultdict from datetime import datetime from typing import Annotated, Any import arcticdb as adb import pandas as pd from fastapi import APIRouter, Path from nicegui import app, ui from aggridx import aggridx from registry import Registry PGroup = Annotated[str, Path(pattern=r"^p\d{5}$")] uri = "lmdb://adb" ac = adb.Arctic(uri) lib = ac.get_library( "stand", create_if_missing=True, library_options=adb.LibraryOptions(dynamic_schema=True) ) router = APIRouter() grids = defaultdict(Registry) OPTIONS = { ":getRowId": "(params) => params.data.index", # set row ID to index column "context": {"pgroup": None}, "defaultColDef": { "filter": True, "editable": True, "sortable": True, "resizable": True }, "pagination": True, "paginationAutoPageSize": True, "theme": "balham" } def update_adb(evt): if evt.args.get("source") != "edit": # ignore event if it was no direct edit return pgroup = evt.args["context"]["pgroup"] row_id = evt.args["rowId"] col_id = evt.args["colId"] new_val = evt.args["newValue"] index = datetime.fromisoformat(row_id) # nicegui converts datetime to str df = lib.read(pgroup, date_range=[index]).data df.at[index, col_id] = new_val lib.update(pgroup, df) def update_grids(evt): sender = evt.sender pgroup = evt.args["context"]["pgroup"] row_index = evt.args["rowIndex"] row_id = evt.args["rowId"] col_id = evt.args["colId"] new_val = evt.args["newValue"] for grid in grids[pgroup]: grid.set_cell_server(row_index, col_id, new_val) if grid == sender: # the sender is already up-to-date continue grid.set_cell_client(row_id, col_id, new_val) @ui.page("/{pgroup}") def table(pgroup: PGroup): # with ui.left_drawer(value=False) as ld: # dark = ui.dark_mode(value=True) # ui.switch("dark mode").bind_value(dark) # with ui.page_sticky(position="left", x_offset=-12.5): # def cb(): # ld.toggle() # btn.icon = "sym_o_left_panel_close" if ld.value else "sym_o_left_panel_open" # btn = ui.button(icon="sym_o_left_panel_open", on_click=cb).props("flat dense") try: df = lib.read(pgroup).data except adb.exceptions.NoSuchVersionException: df = pd.DataFrame() df = df.reset_index() options = OPTIONS.copy() options["context"]["pgroup"] = pgroup grid = aggridx.from_pandas(df, options=options) grid.classes("h-[calc(100vh-2rem)]") # full height minus padding grid.on("cellValueChanged", update_adb) grid.on("cellValueChanged", update_grids) grids[pgroup].add(grid) @router.post("/{pgroup}/append") def append(pgroup: PGroup, row: dict[str, Any]): now = datetime.now() df = pd.DataFrame(row, index=[now]) lib.append(pgroup, df) now = str(now) # nicegui converts datetime to str row = {"index": now, **row} # setdefault would not force index to be the first column res = [] for grid in grids[pgroup]: grid.append(row) res.append(grid.options) return res app.include_router(router) ui.run(title="stand", favicon="favicon.png", fastapi_docs=True, dark=True, show=False)