143 lines
No EOL
5 KiB
Python
143 lines
No EOL
5 KiB
Python
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 = '''
|
|
<a rel="license" href="https://www.gnu.org/licenses/gpl.html">
|
|
<img alt="GNU GPL v3 license"style="border-width:0"
|
|
src="/static/icons/gplv3-88x31.png" title="GPL Open Source license"/>
|
|
</a>
|
|
'''
|
|
|
|
default_content = '''
|
|
Gisaf is free, open source software for geomatics and GIS:
|
|
<a href="http://redmine.auroville.org.in/projects/gisaf">Gisaf</a>.
|
|
'''
|
|
|
|
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 |