import logging from pathlib import Path from fastapi import Depends, APIRouter, HTTPException, status, responses from sqlalchemy.orm import selectinload from sqlmodel import select import pandas as pd import geopandas as gpd from gisaf.config import conf from gisaf.utils import dict_array_to_list from gisaf.database import fastapi_db_session as db_session from gisaf.models.authentication import User from gisaf.models.dashboard import ( DashboardPage, DashboardPageSection, DashboardPageMetaData, DashboardGroup, DashboardHome, Dashboard, DashboardSection ) from gisaf.models.misc import NotADataframeError from gisaf.security import get_current_active_user logger = logging.getLogger(__name__) default_footer = ''' GNU GPL v3 license ''' default_content = ''' Gisaf is free, open source software for geomatics and GIS: Gisaf. ''' api = APIRouter( tags=["dashboard"], # dependencies=[Depends(get_token_header)], responses={404: {"description": "Not found"}}, ) @api.get('/groups') async def get_groups( db_session: db_session, ) -> list[DashboardGroup]: query = select(DashboardPage) data = await db_session.exec(query) groups: dict[str, DashboardPageMetaData] = {} for page in data.all(): page_field = DashboardPageMetaData(name=page.name, group=page.group, description=page.description, viewable_role=page.viewable_role ) group = groups.get(page.group) if group is None: group = DashboardGroup(name=page.group, pages=[page_field]) groups[page.group] = group else: group.pages.append(page_field) return groups.values() @api.get('/home') async def get_home() -> DashboardHome: content_path = Path(conf.gisaf.dashboard_home.content_file).expanduser() footer_path = Path(conf.gisaf.dashboard_home.footer_file).expanduser() if content_path.is_file(): content = content_path.read_text() else: content = default_content if footer_path.is_file(): footer = footer_path.read_text() else: footer = default_footer return DashboardHome( title=conf.gisaf.dashboard_home.title, content=content, footer=footer, ) @api.get('/page/{group}/{name}') async def get_dashboard_page(group: str, name: str, db_session: db_session, user: User = Depends(get_current_active_user), ) -> Dashboard: query1 = select(DashboardPage).\ options(selectinload(DashboardPage.sections)).\ where((DashboardPage.name==name) & (DashboardPage.group==group)) data1 = await db_session.exec(query1) page = data1.one_or_none() if not page: raise HTTPException(status.HTTP_404_NOT_FOUND) query2 = select(DashboardPageSection)\ .where(DashboardPageSection.dashboard_page_id==page.id)\ .options(selectinload(DashboardPageSection.dashboard_page))\ .order_by(DashboardPageSection.name) data2 = await db_session.exec(query2) sections = data2.all() if page.viewable_role: if not(user and user.has_role(page.viewable_role)): username = user.username if user is not None else "Anonymous" logger.info(f'{username} tried to access dashboard page {name}') raise HTTPException(status.HTTP_401_UNAUTHORIZED) dp = Dashboard( name=page.name, group=page.group, description=page.description, html=page.html, time=page.time, notebook=page.get_notebook_url(), attachment=page.get_attachment_url(), expandedPanes=[ p.strip() for p in page.expanded_panes.split(',') ] if page.expanded_panes else [], sections=[ DashboardSection( name=dps.name, plot=dps.get_plot_url() ) for dps in sections ] ) try: df = page.get_page_df() if df is not None: ## TODO: plot as external file, like for sections ## Convert Geopandas dataframe to Pandas if isinstance(df, gpd.GeoDataFrame): gdf = pd.DataFrame(df.drop(columns=['geometry'])) df = gdf dp.dfData = df.reset_index().to_dict(orient='records') except NotADataframeError: logger.warning(f'Dashboard: cannot read dataframe for page {page.name}') except Exception as err: logger.warning(f'Dashboard: cannot add dataframe for page {page.name}, ' 'see debug message') logger.exception(err) if page.plot: plot = page.get_plot() ## Convert manually numpy arrays to lists dp.plotData = [dict_array_to_list(d.to_plotly_json()) for d in plot.data] dp.plotLayout = plot.layout.to_plotly_json() return dp