147 lines
6.2 KiB
Python
147 lines
6.2 KiB
Python
"""
|
|
TOML interface
|
|
--------------
|
|
|
|
Remember, that keys with dots have to be quoted.
|
|
Apart from improting projects yourdelf with python scripts, this package also allows for
|
|
the import of projects via TOML.
|
|
"""
|
|
|
|
|
|
import tomllib as toml
|
|
import shutil
|
|
from .input import sfcf, openQCD
|
|
from .main import import_project, update_aliases
|
|
from .meas_io import write_measurement
|
|
import datalad.api as dl
|
|
import os
|
|
from .input.implementations import codes as known_codes
|
|
|
|
|
|
def check_project_data(d: dict) -> None:
|
|
if 'project' not in d.keys() or 'measurements' not in d.keys() or len(list(d.keys())) > 2:
|
|
raise ValueError('There should only be two key on the top level, "project" and "measurements"!')
|
|
project_data = d['project']
|
|
if 'url' not in project_data.keys():
|
|
raise ValueError('project.url is missing!')
|
|
if 'code' not in project_data.keys():
|
|
raise ValueError('project.code is missing!')
|
|
if 'measurements' not in d.keys():
|
|
raise ValueError('No measurements to import!')
|
|
return
|
|
|
|
|
|
def check_measurement_data(measurements: dict, code: str) -> None:
|
|
var_names: list[str] = []
|
|
if code == "sfcf":
|
|
var_names = ["path", "ensemble", "param_file", "version", "prefix", "cfg_seperator", "names"]
|
|
elif code == "openQCD":
|
|
var_names = ["path", "ensemble", "measurement", "prefix"] # , "param_file"
|
|
for mname, md in measurements.items():
|
|
for var_name in var_names:
|
|
if var_name not in md.keys():
|
|
raise ImportError("Measurment '" + mname + "' does not possess nessecary variable '" + var_name + "'. \
|
|
Please add this to the measurements definition.")
|
|
return
|
|
|
|
|
|
def import_tomls(path: str, files: str, copy_files: bool=True) -> None:
|
|
for file in files:
|
|
import_toml(path, file, copy_files)
|
|
|
|
|
|
def import_toml(path: str, file: str, copy_file: bool=True) -> None:
|
|
"""
|
|
Import a project decribed by a .toml file.
|
|
|
|
Parameters
|
|
----------
|
|
path: str
|
|
Path to the backlog directory.
|
|
file: str
|
|
Path to the description file.
|
|
"""
|
|
print("Import project as decribed in " + file)
|
|
with open(file, 'rb') as fp:
|
|
toml_dict = toml.load(fp)
|
|
check_project_data(toml_dict)
|
|
project: dict = toml_dict['project']
|
|
if project['code'] not in known_codes:
|
|
raise ValueError('Code' + project['code'] + 'has no import implementation!')
|
|
measurements: dict = toml_dict['measurements']
|
|
check_measurement_data(measurements, project['code'])
|
|
aliases = project.get('aliases', None)
|
|
uuid = project.get('uuid', None)
|
|
if uuid is not None:
|
|
if not os.path.exists(path + "/projects/" + uuid):
|
|
uuid = import_project(path, project['url'], aliases=aliases)
|
|
else:
|
|
update_aliases(path, uuid, aliases)
|
|
else:
|
|
uuid = import_project(path, project['url'], aliases=aliases)
|
|
for mname, md in measurements.items():
|
|
print("Import measurement: " + mname)
|
|
ensemble = md['ensemble']
|
|
if project['code'] == 'sfcf':
|
|
param = sfcf.read_param(path, uuid, md['param_file'])
|
|
if 'names' in md.keys():
|
|
measurement = sfcf.read_data(path, uuid, md['path'], md['prefix'], param,
|
|
version=md['version'], cfg_seperator=md['cfg_seperator'], sep='/', names=md['names'])
|
|
else:
|
|
measurement = sfcf.read_data(path, uuid, md['path'], md['prefix'], param,
|
|
version=md['version'], cfg_seperator=md['cfg_seperator'], sep='/')
|
|
print(mname + " imported.")
|
|
elif project['code'] == 'openQCD':
|
|
if md['measurement'] == 'ms1':
|
|
param = openQCD.read_ms1_param(path, uuid, md['param_file'])
|
|
param['type'] = 'ms1'
|
|
measurement = openQCD.read_rwms(path, uuid, md['path'], param, md["prefix"], version=md["version"], names=md['names'], files=md['files'])
|
|
elif md['measurement'] == 't0':
|
|
if 'param_file' in md:
|
|
param = openQCD.read_ms3_param(path, uuid, md['param_file'])
|
|
else:
|
|
param = {}
|
|
for rwp in ["integrator", "eps", "ntot", "dnms"]:
|
|
param[rwp] = "Unknown"
|
|
param['type'] = 't0'
|
|
measurement = openQCD.extract_t0(path, uuid, md['path'], param, md["prefix"], md["dtr_read"], md["xmin"], md["spatial_extent"], fit_range=md.get('fit_range', 5), postfix=md.get('postfix', None), names=md.get('names', None))
|
|
elif md['measurement'] == 't1':
|
|
if 'param_file' in md:
|
|
param = openQCD.read_ms3_param(path, uuid, md['param_file'])
|
|
param['type'] = 't1'
|
|
measurement = openQCD.extract_t1(path, uuid, md['path'], param, md["prefix"], md["dtr_read"], md["xmin"], md["spatial_extent"], fit_range=md.get('fit_range', 5), postfix=md.get('postfix', None), names=md.get('names', None))
|
|
|
|
write_measurement(path, ensemble, measurement, uuid, project['code'], (md['param_file'] if 'param_file' in md else None))
|
|
|
|
if not os.path.exists(os.path.join(path, "toml_imports", uuid)):
|
|
os.makedirs(os.path.join(path, "toml_imports", uuid))
|
|
if copy_file:
|
|
import_file = os.path.join(path, "toml_imports", uuid, file.split("/")[-1])
|
|
shutil.copy(file, import_file)
|
|
dl.save(import_file, message="Import using " + import_file, dataset=path)
|
|
print("File copied to " + import_file)
|
|
print("Imported project.")
|
|
return
|
|
|
|
|
|
def reimport_project(path, uuid):
|
|
"""
|
|
Reimport an existing project using the files that are already available for this project.
|
|
|
|
Parameters
|
|
----------
|
|
path: str
|
|
Path to repository
|
|
uuid: str
|
|
uuid of the project that is to be reimported.
|
|
"""
|
|
config_path = "/".join([path, "import_scripts", uuid])
|
|
for p, filenames, dirnames in os.walk(config_path):
|
|
for fname in filenames:
|
|
import_toml(path, os.path.join(config_path, fname), copy_file=False)
|
|
return
|
|
|
|
|
|
def update_project(path, uuid):
|
|
dl.update(how='merge', follow='sibling', dataset=os.path.join(path, "projects", uuid))
|
|
# reimport_project(path, uuid)
|