185 lines
7.9 KiB
Python
185 lines
7.9 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
|
|
|
|
import datalad.api as dl
|
|
from .tracker import save
|
|
from .input import sfcf, openQCD
|
|
from .main import import_project, update_aliases
|
|
from .meas_io import write_measurement
|
|
import os
|
|
from .input.implementations import codes as known_codes
|
|
from typing import Any
|
|
|
|
|
|
def replace_string(string: str, name: str, val: str) -> str:
|
|
if '{' + name + '}' in string:
|
|
n = string.replace('{' + name + '}', val)
|
|
return n
|
|
else:
|
|
return string
|
|
|
|
|
|
def replace_in_meas(measurements: dict[str, dict[str, Any]], vars: dict[str, str]) -> dict[str, dict[str, Any]]:
|
|
# replace global variables
|
|
for name, value in vars.items():
|
|
for m in measurements.keys():
|
|
for key in measurements[m].keys():
|
|
if isinstance(measurements[m][key], str):
|
|
measurements[m][key] = replace_string(measurements[m][key], name, value)
|
|
elif isinstance(measurements[m][key], list):
|
|
for i in range(len(measurements[m][key])):
|
|
measurements[m][key][i] = replace_string(measurements[m][key][i], name, value)
|
|
return measurements
|
|
|
|
|
|
def fill_cons(measurements: dict[str, dict[str, Any]], constants: dict[str, str]) -> dict[str, dict[str, Any]]:
|
|
for m in measurements.keys():
|
|
for name, val in constants.items():
|
|
if name not in measurements[m].keys():
|
|
measurements[m][name] = val
|
|
return measurements
|
|
|
|
|
|
def check_project_data(d: dict[str, dict[str, str]]) -> None:
|
|
if 'project' not in d.keys() or 'measurements' not in d.keys() or len(list(d.keys())) > 4:
|
|
raise ValueError('There should only be maximally be four keys on the top level, "project" and "measurements" are mandatory, "contants" is optional!')
|
|
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[str, dict[str, str]], 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: list[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[str, Any] = toml_dict['project']
|
|
if project['code'] not in known_codes:
|
|
raise ValueError('Code' + project['code'] + 'has no import implementation!')
|
|
measurements: dict[str, dict[str, Any]] = toml_dict['measurements']
|
|
measurements = fill_cons(measurements, toml_dict['constants'] if 'constants' in toml_dict else {})
|
|
measurements = replace_in_meas(measurements, toml_dict['replace'] if 'replace' in toml_dict else {})
|
|
check_measurement_data(measurements, project['code'])
|
|
aliases = project.get('aliases', [])
|
|
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, str(md["prefix"]), int(md["dtr_read"]), int(md["xmin"]), int(md["spatial_extent"]),
|
|
fit_range=int(md.get('fit_range', 5)), postfix=str(md.get('postfix', '')), names=md.get('names', []), files=md.get('files', []))
|
|
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, str(md["prefix"]), int(md["dtr_read"]), int(md["xmin"]), int(md["spatial_extent"]),
|
|
fit_range=int(md.get('fit_range', 5)), postfix=str(md.get('postfix', '')), names=md.get('names', []), files=md.get('files', []))
|
|
|
|
write_measurement(path, ensemble, measurement, uuid, project['code'], (md['param_file'] if 'param_file' in md else ''))
|
|
|
|
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)
|
|
save(path, files=[import_file], message="Import using " + import_file)
|
|
print("File copied to " + import_file)
|
|
print("Imported project.")
|
|
return
|
|
|
|
|
|
def reimport_project(path: str, uuid: str) -> None:
|
|
"""
|
|
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: str, uuid: str) -> None:
|
|
dl.update(how='merge', follow='sibling', dataset=os.path.join(path, "projects", uuid))
|
|
# reimport_project(path, uuid)
|
|
return
|