#!/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 auth.login import router as login_router from auth.logout import logout from auth.mw import AuthMiddleware from auth.secret import get_secret from aggridx import aggridx from registry import Registry app.include_router(login_router) app.add_middleware(AuthMiddleware) 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("/") def main(): with ui.column().classes("absolute-center items-center"): username = app.storage.user.get("username", "unknown user") ui.label(f"Hello {username}!").classes("text-2xl") ui.button("log out", icon="logout", on_click=logout) pgroups = app.storage.user.get("pgroups", set()) ui.select( label="pgroup", options=sorted(pgroups), with_input=True, on_change=lambda e: ui.navigate.to(f"/tables/{e.value}") ) @ui.page("/tables/{pgroup}") def table(pgroup: PGroup): pgroups = app.storage.user.get("pgroups", set()) if pgroup in pgroups: table_show(pgroup) else: table_deny(pgroup) def table_show(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) def table_deny(pgroup): with ui.column().classes("absolute-center items-center gap-8"): ui.icon("sym_o_front_hand", size="xl") ui.label(f"access to {pgroup} denied").classes("text-2xl") @router.post("/tables/{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", storage_secret=get_secret(), fastapi_docs=True, dark=True, show=False)