Source code for solidipes_core_plugin.uploaders.dtool
import os
import dtoolcore
from dtoolcore import DataSet, DtoolCoreTypeError, ProtoDataSet
from solidipes.loaders.file import load_file
from solidipes.uploaders.uploader import Uploader
from solidipes.utils import generate_readme, include_metadata_description
from solidipes.utils import study_medatada_mandatory_fields as mandatory_fields
from solidipes.utils import study_medatada_removed_fields_upload as removed_fields
from solidipes.utils.utils import classproperty, optional_parameter, parameter
################################################################
[docs]
class DToolUploader(Uploader):
parser_key = "dtool"
command_help = "Publish study to dtool"
name = "DTool"
@classproperty
def report_widget_class(self):
from solidipes_core_plugin.reports.widgets.dtool import DToolPublish
return DToolPublish
@parameter
def url() -> str:
"""URI where to save the repository"""
pass
@optional_parameter
def dataset_name() -> str:
"""name to give to the dataset if not existing"""
return "Unsetname"
@optional_parameter
def freeze() -> bool:
"""requests to freeze the dataset after upload"""
return False
@optional_parameter
def no_metadata() -> bool:
"""requests to not export metadata to dtool"""
return False
################################################################
[docs]
def main(args) -> None:
"""Upload content to a DTool repository."""
from solidipes.utils.utils import get_study_root_path
print(args)
if args.directory is None:
args.root_directory = get_study_root_path()
else:
args.root_directory = args.directory
# Zip directory into temporary file
generate_readme()
file_list = create_file_list(args)
metadata = load_and_check_metadata(args)
print("Uploading archive")
upload_dtool_dataset(
args.url, file_list, metadata, freeze_flag=args.freeze, name=args.dataset_name, no_metadata=args.no_metadata
)
################################################################
[docs]
def create_file_list(config):
from solidipes.scanners.scanner_local import ExportScanner
scanner = ExportScanner()
filepaths = scanner.get_filepath_list()
return filepaths
################################################################
[docs]
def load_and_check_metadata(config):
"""Load/create metadata file and check if mandatory fields are present"""
dir_path = config.root_directory
from solidipes.utils import get_study_metadata, get_study_metadata_path
metadata = get_study_metadata(initial_path=dir_path, check_existence=True)
metadata_path = get_study_metadata_path(initial_path=dir_path)
# Replace description with content from DESCRIPTION.md converted in HTML
metadata = include_metadata_description(metadata, md_to_html=True, use_readme=False, initial_path=dir_path)
print(metadata, mandatory_fields.keys())
# Check if mandatory fields are present
for field in mandatory_fields.keys():
if field not in metadata or not metadata[field]:
raise ValueError(
f'Error: field "{field}" is missing from metadata file or is empty. Please edit {metadata_path} and try'
" again."
)
# Check that creators is a list
if not isinstance(metadata["creators"], list):
raise ValueError(f'Error: field "creators" must be a list. Please edit {metadata_path} and try again.')
# Check that each creator has a name
for creator in metadata["creators"]:
if "name" not in creator or not creator["name"]:
raise ValueError(
f'Error: field "name" is missing from one of the creators. Please edit {metadata_path} and try again.'
)
# Clean
for field in removed_fields:
if field in metadata:
del metadata[field]
if "related_identifiers" in metadata:
for related_identifier in metadata["related_identifiers"]:
if "relation" in related_identifier and related_identifier["relation"] == "isVersionOf":
related_identifier["relation"] = "isNewVersionOf"
return metadata
################################################################
[docs]
def upload_dtool_dataset(uri, filenames, metadata, name=None, freeze_flag=False, no_metadata=False):
from solidipes.utils.progress import get_progress_bar
user_metadata = metadata
print(metadata)
creator_username = metadata["creators"][0]["username"]
print(creator_username)
base_uri = dtoolcore.utils.sanitise_uri(uri)
try:
proto_ds = ProtoDataSet.from_uri(base_uri)
proto_dataset = True
except (DtoolCoreTypeError, Exception):
proto_dataset = False
try:
DataSet.from_uri(base_uri)
frozen_dataset = True
except (DtoolCoreTypeError, Exception):
frozen_dataset = False
dataset_exists = frozen_dataset or proto_dataset
if dataset_exists:
print(f"Dataset already exists at {base_uri}.")
else:
print(f"Creating new dataset at {base_uri}...")
base_uri = base_uri.split(name)[0]
proto_ds = dtoolcore.create_proto_dataset(
name=name, base_uri=base_uri, readme_content=metadata["description"], creator_username=creator_username
)
if frozen_dataset:
print("Frozen dataset: abort")
return
import yaml
print("Setting global metadata as readme...")
proto_ds.put_readme(yaml.safe_dump(user_metadata))
loaders = []
with get_progress_bar("Uploading/Updating files", total=len(filenames)) as progress:
for filepath in filenames:
progress.update(advance=1, text=filepath)
if not os.path.exists(filepath):
print(f"Warning: File {filepath} not found. Skipping.")
continue
f = load_file(filepath)
proto_ds.put_item(filepath, filepath)
from dtoolcore.utils import generate_identifier
loaders.append((f, generate_identifier(f.path)))
from solidipes.loaders.rocrate_metadata import rocrate_metadata
metadata_list = set()
with get_progress_bar("Listing metadata of files", total=len(loaders)) as progress:
for f, _id in loaders:
progress.update(advance=1, text=f.path)
metadata = f.get_cached_metadata()
for key, value in metadata.items():
attribute = getattr(f.__class__, key)
if not isinstance(attribute, rocrate_metadata):
continue
metadata_list.add(key)
if no_metadata is False:
with get_progress_bar("Adding metadata", total=len(loaders)) as progress:
for f, _id in loaders:
progress.update(advance=1, text=f.path)
metadata = f.get_cached_metadata()
for key in metadata_list:
value = metadata.get(key, None)
# print("Adding metadata", f.path, _id, key, value)
proto_ds.add_item_metadata(f.path, key, str(value))
_ids = list(proto_ds._identifiers())
for f, _id in loaders:
if _id not in _ids:
raise RuntimeError(f"did not register {f.path}: {_id} ?")
if freeze_flag:
print("Freezing dataset...")
proto_ds.freeze()
print(f"Dataset finalized and frozen at {uri}.")
else:
print(f"Dataset updated as a draft (ProtoDataSet) at {uri}.")