jkuhl/corrlib
1
0
Fork 0
Code Issues 11 Pull requests 1 Releases Packages Activity Actions

correct mypy issues

Browse source
This commit is contained in:
Justus Kuhlmann 2025-12-02 12:35:09 +01:00
commit 4546688d97
Signed by: jkuhl
GPG key ID: 00ED992DD79B85A6
8 changed files with 64 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

32
corrlib/toml.py
View file

@ -16,15 +16,16 @@ from .meas_io import write_measurement
import datalad.api as dl
import os
from .input.implementations import codes as known_codes
from typing import Any
def replace_string(string: str, name: str, val: str):
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, vars: dict[str, str]):
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():
@ -36,7 +37,7 @@ def replace_in_meas(measurements: dict, vars: dict[str, str]):
measurements[m][key][i] = replace_string(measurements[m][key][i], name, value)
return measurements
def fill_cons(measurements, constants):
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():
@ -44,7 +45,7 @@ def fill_cons(measurements, constants):
return measurements
def check_project_data(d: dict) -> None:
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']
@ -57,7 +58,7 @@ def check_project_data(d: dict) -> None:
return
def check_measurement_data(measurements: dict, code: str) -> None:
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"]
@ -91,14 +92,14 @@ def import_toml(path: str, file: str, copy_file: bool=True) -> None:
with open(file, 'rb') as fp:
toml_dict = toml.load(fp)
check_project_data(toml_dict)
project: dict = toml_dict['project']
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 = toml_dict['measurements']
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', None)
aliases = project.get('aliases', [])
uuid = project.get('uuid', None)
if uuid is not None:
if not os.path.exists(path + "/projects/" + uuid):
@ -133,16 +134,16 @@ def import_toml(path: str, file: str, copy_file: bool=True) -> None:
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), files=md.get('files', None))
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, 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), files=md.get('files', None))
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 None))
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))
@ -155,7 +156,7 @@ def import_toml(path: str, file: str, copy_file: bool=True) -> None:
return
def reimport_project(path, uuid):
def reimport_project(path: str, uuid: str) -> None:
"""
Reimport an existing project using the files that are already available for this project.
@ -173,6 +174,7 @@ def reimport_project(path, uuid):
return
def update_project(path, uuid):
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