#!/bin/env python
################################################################
import argparse
import os
from typing import TYPE_CHECKING
from solidipes.reports.report import Report
from solidipes.reports.widgets.solidipes_buttons import SolidipesButtons as SPB
from solidipes.reports.widgets.utils import FileWrapper, transform_to_subtree
from solidipes.scanners.scanner import Scanner, StreamlitProgressBar, list_files
from solidipes.utils import add_completed_stage, is_stage_completed, logging, remove_completed_stage
from solidipes.utils.git_infos import GitInfos
################################################################
if TYPE_CHECKING:
import streamlit as st
else:
import lazy_loader as lazy
st = lazy.load("streamlit")
print = logging.invalidPrint
logger = logging.getLogger()
################################################################
WebReport_pages = []
[docs]
def sp_page(foo):
WebReport_pages.append(foo.__name__)
return foo
[docs]
class WebReport:
def __init__(self):
self.git_infos = GitInfos()
self.display_push_button = False
self.file_wildcard = "*"
self.file_error_checkbox = None
self.scanner = Scanner()
st.set_page_config(
layout="wide",
page_icon="https://gitlab.com/dcsm/website/-/raw/main/static/favicon.ico",
# initial_sidebar_state="collapsed",
)
if "currently_opened" not in st.session_state:
st.session_state["currently_opened"] = None
[docs]
def createLayouts(self):
from solidipes.reports.widgets.gitlab_issues import GitlabIssues
from solidipes.reports.widgets.solidipes_logo_widget import SolidipesLogoWidget
from solidipes.reports.widgets.zenodo import ZenodoInfos, ZenodoPublish
self.progress_layout = st.empty()
SolidipesLogoWidget(layout=st.sidebar, short=True, width="25%")
st.sidebar.markdown("---")
self.info_layout = st.sidebar.container()
st.sidebar.markdown("---")
self.gitlab_control = st.sidebar.container()
self.jupyter_control = st.sidebar.container()
self.filebrowser_control = st.sidebar.container()
self.update_buttons = st.sidebar.container()
self.file_selector = st.container()
self.path_selector = st.sidebar.container()
if self.git_infos.repository is not None:
self.git_control = st.sidebar.container()
self.env_layout = st.sidebar.container()
self.options = st.sidebar.expander("Options")
self.main_layout = st.container()
self.file_layout = st.container()
# self.modified_state = self.main_layout.empty()
self.global_message = self.main_layout.container()
self.header_layout = self.main_layout.container()
self.tab_metadata, self.tab_files = self.main_layout.container(), self.main_layout.container()
self.zenodo_publish = ZenodoPublish(self.tab_metadata, self.global_message, self.progress_layout)
self.zenodo_infos = ZenodoInfos(self.tab_metadata)
if self.git_infos.origin is not None:
self.gitlab_issues = GitlabIssues(self.main_layout)
self.files_container = self.main_layout.container()
self.logs = self.main_layout.container()
[docs]
def scan_directories(self, dir_path):
all_paths = []
nodes = None
with st.spinner("Loading directories..."):
if "scanned_files" not in st.session_state:
st.session_state["scanned_files"] = {}
h = self.scanner.get_dirpath_tree()
s_files = st.session_state["scanned_files"]
s_files["all_paths"] = self.scanner.get_path_list()
s_files["nodes"] = transform_to_subtree(h)
else:
s_files = st.session_state["scanned_files"]
nodes = s_files["nodes"]
all_paths = s_files["all_paths"]
return all_paths, nodes
[docs]
def step_bar(self, page):
from solidipes.reports.widgets.step_bar import StepBar
if page == "display_page":
page = "curation"
StepBar(current_step=page)
st.markdown("---\n\n")
[docs]
def main(self, dir_path):
self.dir_path = dir_path
if "page" in st.query_params:
page = st.query_params["page"]
if page not in WebReport_pages:
st.error(f"Invalid page '{page}'")
return
try:
self.step_bar(page)
page_method = getattr(self, page)
except AttributeError:
st.error(f"Invalid page '{page}'")
return
return page_method()
return self.main_page()
[docs]
@sp_page
def display_page(self):
from solidipes.reports.widgets.display_file import DisplayFile
self.createLayouts()
self.info_layout.write(
"This page shows you the validation state and a visualization of one element of your dataset. You can use"
" the “Discussions” to make any relevant comments ; this will cause the file to be tagged as erroneous"
" until the issue is resolved."
)
if "file" not in st.query_params:
st.error(f"Wrong url {[k+'='+v for k, v in st.query_params.items()]}")
return
fname = st.query_params["file"]
loader = st.query_params.get("loader", "")
paths = st.query_params.get("paths", [])
return DisplayFile(filename=fname, loader_name=loader, paths_str=paths, layout=self.file_layout)
[docs]
def main_page(self):
from solidipes.reports.widgets.front_page import FrontPage
FrontPage()
[docs]
@sp_page
def export(self):
self.createLayouts()
self.info_layout.write(
"*This is the final step, which allows you to publish your curated dataset to an online archive. **Please"
" check** that the exported archive contains all the files you wish to be published. Note that some files"
" are added inside a .solidipes directory for forward compatibility.*"
)
self.zenodo_publish.show()
[docs]
@sp_page
def acquisition(self):
self.createLayouts()
self.info_layout.write(
"*Here you can acquire and organise all the files you wish to publish. Much like a regular file browser, it"
" allows you to view and upload all files and folders in your project directory.*"
)
if is_stage_completed(0):
self.info_layout.write(
"*This stage has been validated. If you still need to make changes, you can **invalidate the stage**"
" with the button below.*"
)
self.info_layout.button(
"Invalidate acquisition",
on_click=lambda: remove_completed_stage(0),
use_container_width=True,
type="primary",
)
else:
self.info_layout.write("*Once ready you can **validate the acquisition** with the button below.*")
SPB(layout=self.info_layout)._link_button(
"Validate acquisition",
"?page=curation",
action=lambda: add_completed_stage(0),
use_container_width=True,
type="primary",
)
SPB()._iframe_filebrowser(self.dir_path)
[docs]
@sp_page
def curation(self):
self.createLayouts()
self.info_layout.write(
"*On this page, you can view all files in your dataset, as well as their validation state.*"
)
self.info_layout.write("*For an in-depth view of a given file, you can click on the “View File” button.*")
self.info_layout.write("**Each file** *marked* 🚫 *needs to be checked to complete curation stage.*")
self.info_layout.write("To change the extension of a file, double-click in the “extension” column.")
self.scanner.root_path = self.dir_path
if "GUI_files" not in st.session_state:
st.session_state["GUI_files"] = {}
self.show_advanced = False
SPB(layout=self.update_buttons)._link_button(
"Next (metadata)",
"?page=metadata",
use_container_width=True,
type="primary",
)
SPB(layout=self.gitlab_control)._open_in_gitlab_button(self.git_infos.origin)
# self.show_advanced = self.options.checkbox("Advanced", value=False)
if self.show_advanced:
self._environment_info()
if self.display_push_button:
from solidipes.reports.widgets.git import GIT
GIT(container_infos=self.git_infos, container_state=self.modified_state)
if self.git_infos.origin is not None:
self.gitlab_issues.show()
SPB(layout=self.jupyter_control)._open_in_jupyterlab_button()
SPB(layout=self.filebrowser_control)._open_in_filebrowser_button()
if SPB(layout=self.file_selector)._force_rescan_button():
clear_session_state()
all_paths, nodes = self.scan_directories(self.dir_path)
if "all_found_files" not in st.session_state:
self.scanner.progress_bar = StreamlitProgressBar("Loading files", self.progress_layout)
found = self.scanner.get_filtered_loader_tree([p for p in all_paths], recursive=False)
files = list_files(found)
files_dict = self.scanner.get_filtered_loader_dict([p for p in all_paths], recursive=False)
files_dict = {k: FileWrapper(v) for k, v in files_dict.items()}
st.session_state["all_found_files"] = files
st.session_state["all_found_files_dict"] = files_dict
all_found_files = st.session_state["all_found_files"]
if not all_found_files:
# st.markdown(f"#### Nothing in the paths: {all_paths}")
st.markdown("#### Nothing in the filtered files")
return
with self.tab_files:
from solidipes.reports.widgets.file_list import FileList
FileList(all_found_files=all_found_files, progress_layout=self.progress_layout)
with self.progress_layout:
with st.spinner("Saving cache to YAML format"):
from solidipes.loaders.cached_metadata import CachedMetadata
if CachedMetadata._global_cached_metadata is not None:
CachedMetadata._write_cached_metadata_to_yaml()
if self.show_advanced:
self.logs.markdown("---")
from solidipes.reports.widgets.solidipes_logs import SolidipesLogs
SolidipesLogs(layout=self.logs)
[docs]
def _environment_info(self):
with self.env_layout.expander("Environment"):
st.write("sh env")
table_env = [k for k in os.environ.items()]
st.dataframe(table_env, use_container_width=True)
import pkg_resources
st.write("pip packages")
table_env = [p.project_name for p in pkg_resources.working_set]
st.dataframe(table_env, use_container_width=True)
################################################################
[docs]
def clear_session_state():
logger.info("Clearing session state")
keys = [k for k in st.session_state]
for k in keys:
del st.session_state[k]
from solidipes.loaders.cached_metadata import CachedMetadata
CachedMetadata.close_cached_metadata()
################################################################
[docs]
class WebReportSpawner(Report):
command = "web_report"
command_help = "Launch the web graphical interface"
[docs]
def make(self, args: argparse.Namespace):
import subprocess
if args.debug:
os.environ["FULL_SOLIDIPES_LOG"] = "true"
logger.debug(args.additional_arguments)
cmd = f"streamlit run {__file__} {' '.join(args.additional_arguments)}"
logger.warning(cmd)
subprocess.call(cmd, shell=True, cwd=args.dir_path)
[docs]
def populate_arg_parser(self, parser: argparse.ArgumentParser):
parser.add_argument(
"dir_path",
nargs="?",
default=".",
help="Path to the directory to generate the report for. Defaults to current directory",
)
parser.add_argument(
"--debug",
action="store_true",
help="Enable debug mode",
)
parser.add_argument(
"additional_arguments",
nargs=argparse.REMAINDER,
help="Additional arguments to forward to Streamlit",
)
################################################################
if __name__ == "__main__":
from solidipes.utils import logging
logger.info("starting web_report")
web_report = WebReport()
web_report.main("./")