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

finish a first pitch of docstrings
Some checks failed
Mypy / mypy (push) Failing after 2s
Details
Pytest / pytest (3.12) (push) Failing after 2s
Details
Ruff / ruff (push) Failing after 2s
Details
Mypy / mypy (pull_request) Failing after 2s
Details
Pytest / pytest (3.12) (pull_request) Failing after 2s
Details
Pytest / pytest (3.13) (pull_request) Failing after 1s
Details
Pytest / pytest (3.14) (pull_request) Failing after 2s
Details
Pytest / pytest (3.13) (push) Failing after 2s
Details
Pytest / pytest (3.14) (push) Failing after 2s
Details
Ruff / ruff (pull_request) Failing after 2s
Details

Browse source
This commit is contained in:
Justus Kuhlmann 2026-02-12 18:39:04 +01:00
commit 4631769e81
Signed by: jkuhl
GPG key ID: 00ED992DD79B85A6
7 changed files with 502 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

256
corrlib/find.py
View file

@ -8,11 +8,23 @@ from .tools import k2m, get_db_file
from .tracker import get
from typing import Any, Optional
# this will implement the search functionality
def _project_lookup_by_alias(db: str, alias: str) -> str:
# this will lookup the project name based on the alias
"""
Lookup a projects UUID by its (human-readable) alias.
Parameters
----------
db: str
The database to look up the project.
alias: str
The alias to look up.
Returns
-------
uuid: str
The UUID of the project with the given alias.
"""
conn = sqlite3.connect(db)
c = conn.cursor()
c.execute(f"SELECT * FROM 'projects' WHERE alias = '{alias}'")
@ -26,6 +38,21 @@ def _project_lookup_by_alias(db: str, alias: str) -> str:
def _project_lookup_by_id(db: str, uuid: str) -> list[tuple[str, str]]:
"""
Return the project information available in the database by UUID.
Parameters
----------
db: str
The database to look up the project.
uuid: str
The uuid of the project in question.
Returns
-------
results: list
The row of the project in the database.
"""
conn = sqlite3.connect(db)
c = conn.cursor()
c.execute(f"SELECT * FROM 'projects' WHERE id = '{uuid}'")
@ -36,6 +63,38 @@ def _project_lookup_by_id(db: str, uuid: str) -> list[tuple[str, str]]:
def _db_lookup(db: str, ensemble: str, correlator_name: str, code: str, project: Optional[str]=None, parameters: Optional[str]=None,
created_before: Optional[str]=None, created_after: Optional[Any]=None, updated_before: Optional[Any]=None, updated_after: Optional[Any]=None) -> pd.DataFrame:
"""
Look up a correlator record in the database by the data given to the method.
Parameters
----------
db: str
The database to look up the record.
ensemble: str
The ensemble the record is associated with.
correlator_name: str
The name of the correlator in question.
code: str
The name of the code which was used to calculate the correlator.
project: str, optional
The UUID of the project the correlator was calculated in.
parameters: str, optional
A dictionary holding the exact parameters for the measurement that are held in the database.
created_before: str, optional
Timestamp string before which the meaurement has been created.
created_after: str, optional
Timestamp string after which the meaurement has been created.
updated_before: str, optional
Timestamp string before which the meaurement has been updated.
updated_after: str, optional
Timestamp string after which the meaurement has been updated.
Returns
-------
results: pd.DataFrame
A pandas DataFrame holding the information received form the DB query.
"""
project_str = project
search_expr = f"SELECT * FROM 'backlogs' WHERE name = '{correlator_name}' AND ensemble = '{ensemble}'"
@ -60,85 +119,112 @@ def _db_lookup(db: str, ensemble: str, correlator_name: str, code: str, project:
def sfcf_filter(results: pd.DataFrame, **kwargs: Any) -> pd.DataFrame:
"""
Filter method for the Database entries holding SFCF calculations.
Parameters
----------
results: pd.DataFrame
The unfiltered pandas DataFrame holding the entries from the database.
offset: list[float], optional
quark_kappas: list[float]
quarks_masses: list[float]
qk1: float, optional
Mass parameter $\kappa_1$ of the first quark.
qk2: float, optional
Mass parameter $\kappa_2$ of the first quark.
qm1: float, optional
Bare quak mass $m_1$ of the first quark.
qm2: float, optional
Bare quak mass $m_1$ of the first quark.
quarks_thetas: list[list[float]], optional
wf1: optional
wf2: optional
Results
-------
results: pd.DataFrame
The filtered DataFrame, only holding the records that fit to the parameters given.
"""
drops = []
for ind in range(len(results)):
result = results.iloc[ind]
if result['code'] == 'sfcf':
param = json.loads(result['parameters'])
if 'offset' in kwargs:
if kwargs.get('offset') != param['offset']:
param = json.loads(result['parameters'])
if 'offset' in kwargs:
if kwargs.get('offset') != param['offset']:
drops.append(ind)
continue
if 'quark_kappas' in kwargs:
kappas = kwargs['quark_kappas']
if (not np.isclose(kappas[0], param['quarks'][0]['mass']) or not np.isclose(kappas[1], param['quarks'][1]['mass'])):
drops.append(ind)
continue
if 'quark_masses' in kwargs:
masses = kwargs['quark_masses']
if (not np.isclose(masses[0], k2m(param['quarks'][0]['mass'])) or not np.isclose(masses[1], k2m(param['quarks'][1]['mass']))):
drops.append(ind)
continue
if 'qk1' in kwargs:
quark_kappa1 = kwargs['qk1']
if not isinstance(quark_kappa1, list):
if (not np.isclose(quark_kappa1, param['quarks'][0]['mass'])):
drops.append(ind)
continue
if 'quark_kappas' in kwargs:
kappas = kwargs['quark_kappas']
if (not np.isclose(kappas[0], param['quarks'][0]['mass']) or not np.isclose(kappas[1], param['quarks'][1]['mass'])):
drops.append(ind)
continue
if 'quark_masses' in kwargs:
masses = kwargs['quark_masses']
if (not np.isclose(masses[0], k2m(param['quarks'][0]['mass'])) or not np.isclose(masses[1], k2m(param['quarks'][1]['mass']))):
drops.append(ind)
continue
if 'qk1' in kwargs:
quark_kappa1 = kwargs['qk1']
if not isinstance(quark_kappa1, list):
if (not np.isclose(quark_kappa1, param['quarks'][0]['mass'])):
else:
if len(quark_kappa1) == 2:
if (quark_kappa1[0] > param['quarks'][0]['mass']) or (quark_kappa1[1] < param['quarks'][0]['mass']):
drops.append(ind)
continue
else:
if len(quark_kappa1) == 2:
if (quark_kappa1[0] > param['quarks'][0]['mass']) or (quark_kappa1[1] < param['quarks'][0]['mass']):
drops.append(ind)
continue
if 'qk2' in kwargs:
quark_kappa2 = kwargs['qk2']
if not isinstance(quark_kappa2, list):
if (not np.isclose(quark_kappa2, param['quarks'][1]['mass'])):
drops.append(ind)
continue
else:
if len(quark_kappa2) == 2:
if (quark_kappa2[0] > param['quarks'][1]['mass']) or (quark_kappa2[1] < param['quarks'][1]['mass']):
drops.append(ind)
continue
if 'qm1' in kwargs:
quark_mass1 = kwargs['qm1']
if not isinstance(quark_mass1, list):
if (not np.isclose(quark_mass1, k2m(param['quarks'][0]['mass']))):
drops.append(ind)
continue
else:
if len(quark_mass1) == 2:
if (quark_mass1[0] > k2m(param['quarks'][0]['mass'])) or (quark_mass1[1] < k2m(param['quarks'][0]['mass'])):
drops.append(ind)
continue
if 'qm2' in kwargs:
quark_mass2 = kwargs['qm2']
if not isinstance(quark_mass2, list):
if (not np.isclose(quark_mass2, k2m(param['quarks'][1]['mass']))):
drops.append(ind)
continue
else:
if len(quark_mass2) == 2:
if (quark_mass2[0] > k2m(param['quarks'][1]['mass'])) or (quark_mass2[1] < k2m(param['quarks'][1]['mass'])):
drops.append(ind)
continue
if 'quark_thetas' in kwargs:
quark_thetas = kwargs['quark_thetas']
if (quark_thetas[0] != param['quarks'][0]['thetas'] and quark_thetas[1] != param['quarks'][1]['thetas']) or (quark_thetas[0] != param['quarks'][1]['thetas'] and quark_thetas[1] != param['quarks'][0]['thetas']):
if 'qk2' in kwargs:
quark_kappa2 = kwargs['qk2']
if not isinstance(quark_kappa2, list):
if (not np.isclose(quark_kappa2, param['quarks'][1]['mass'])):
drops.append(ind)
continue
# careful, this is not save, when multiple contributions are present!
if 'wf1' in kwargs:
wf1 = kwargs['wf1']
if not (np.isclose(wf1[0][0], param['wf1'][0][0], 1e-8) and np.isclose(wf1[0][1][0], param['wf1'][0][1][0], 1e-8) and np.isclose(wf1[0][1][1], param['wf1'][0][1][1], 1e-8)):
else:
if len(quark_kappa2) == 2:
if (quark_kappa2[0] > param['quarks'][1]['mass']) or (quark_kappa2[1] < param['quarks'][1]['mass']):
drops.append(ind)
continue
if 'qm1' in kwargs:
quark_mass1 = kwargs['qm1']
if not isinstance(quark_mass1, list):
if (not np.isclose(quark_mass1, k2m(param['quarks'][0]['mass']))):
drops.append(ind)
continue
if 'wf2' in kwargs:
wf2 = kwargs['wf2']
if not (np.isclose(wf2[0][0], param['wf2'][0][0], 1e-8) and np.isclose(wf2[0][1][0], param['wf2'][0][1][0], 1e-8) and np.isclose(wf2[0][1][1], param['wf2'][0][1][1], 1e-8)):
else:
if len(quark_mass1) == 2:
if (quark_mass1[0] > k2m(param['quarks'][0]['mass'])) or (quark_mass1[1] < k2m(param['quarks'][0]['mass'])):
drops.append(ind)
continue
if 'qm2' in kwargs:
quark_mass2 = kwargs['qm2']
if not isinstance(quark_mass2, list):
if (not np.isclose(quark_mass2, k2m(param['quarks'][1]['mass']))):
drops.append(ind)
continue
else:
if len(quark_mass2) == 2:
if (quark_mass2[0] > k2m(param['quarks'][1]['mass'])) or (quark_mass2[1] < k2m(param['quarks'][1]['mass'])):
drops.append(ind)
continue
if 'quark_thetas' in kwargs:
quark_thetas = kwargs['quark_thetas']
if (quark_thetas[0] != param['quarks'][0]['thetas'] and quark_thetas[1] != param['quarks'][1]['thetas']) or (quark_thetas[0] != param['quarks'][1]['thetas'] and quark_thetas[1] != param['quarks'][0]['thetas']):
drops.append(ind)
continue
# careful, this is not save, when multiple contributions are present!
if 'wf1' in kwargs:
wf1 = kwargs['wf1']
if not (np.isclose(wf1[0][0], param['wf1'][0][0], 1e-8) and np.isclose(wf1[0][1][0], param['wf1'][0][1][0], 1e-8) and np.isclose(wf1[0][1][1], param['wf1'][0][1][1], 1e-8)):
drops.append(ind)
continue
if 'wf2' in kwargs:
wf2 = kwargs['wf2']
if not (np.isclose(wf2[0][0], param['wf2'][0][0], 1e-8) and np.isclose(wf2[0][1][0], param['wf2'][0][1][0], 1e-8) and np.isclose(wf2[0][1][1], param['wf2'][0][1][1], 1e-8)):
drops.append(ind)
continue
return results.drop(drops)
@ -152,17 +238,49 @@ def find_record(path: str, ensemble: str, correlator_name: str, code: str, proje
results = _db_lookup(db, ensemble, correlator_name,code, project, parameters=parameters, created_before=created_before, created_after=created_after, updated_before=updated_before, updated_after=updated_after)
if code == "sfcf":
results = sfcf_filter(results, **kwargs)
elif code == "openQCD":
pass
else:
raise Exception
print("Found " + str(len(results)) + " result" + ("s" if len(results)>1 else ""))
return results.reset_index()
def find_project(path: str, name: str) -> str:
"""
Find a project by it's human readable name.
Parameters
----------
path: str
The path of the library.
name: str
The name of the project to look for in the library.
Returns
-------
uuid: str
The uuid of the project in question.
"""
db_file = get_db_file(path)
get(path, db_file)
return _project_lookup_by_alias(os.path.join(path, db_file), name)
def list_projects(path: str) -> list[tuple[str, str]]:
"""
List all projects known to the library.
Parameters
----------
path: str
The path of the library.
Returns
-------
results: list[Any]
The projects known to the library.
"""
db_file = get_db_file(path)
get(path, db_file)
conn = sqlite3.connect(os.path.join(path, db_file))