Commit c20d2b1e authored by Philipp Middendorf's avatar Philipp Middendorf
Browse files

Merge branch 'saved_attr' into 'master'

update_saved_attributes: allow a different number of records per group

See merge request !32
parents 9dc3d87a 46b38c75
#!/usr/bin/env python3
"""
Create a saved_attributes_new.h5 template using a yaml configuration file (e.g.
saved_attributes_config.yml) and optionally existing records (saved_attributes.h5).
If the yaml config file is not provided, it will dump the existing model to a file
existing_config.yml. If both files are provided, it will update the model based on the
yaml config and keep corresponding records from the h5 file. Records from groups/motors
absent from the yaml config will be discarded.
E.g.:
`python path_to/update_saved_attributes.py --conf saved_attributes_config.yml --file saved_attributes.h5`
"""
import argparse
import copy
......@@ -9,13 +23,15 @@ from typing import Any, Dict, List, Optional
def generate_config(
yaml_config: Optional[Dict[str, Any]], h5_config: Optional[Dict[str, Any]]
yaml_config: Optional[Dict[str, Dict[str, str]]],
h5_config: Optional[Dict[str, List[Dict[str, Any]]]],
) -> Optional[Dict[str, Any]]:
"""
Generate saved attributes based on the new config and existing records.
It uses the keys from the new config for the output model; records related to keys
absent from the new config are discarded.
absent from the new config are discarded. For already existing groups, None records
will be created for the new keys.
:param yaml_config: the new config
:param h5_config: a config extracted from an existing saved_attributes.h5 file
......@@ -30,38 +46,46 @@ def generate_config(
h5_subgroups = list(h5_config.keys())
if len(h5_subgroups) == 0:
raise ValueError("No model defined in h5_config")
nb_records = len(h5_config[h5_subgroups[0]]) - 1 # index 0 which is the metadata
if nb_records == 0:
return yaml_config
# get the number of existing records per group
group_records: Dict[str, int] = {} # e.g. {'/Default': 1}
for subgroup in h5_subgroups:
group = "/" + subgroup.split(sep="/", maxsplit=2)[1]
if not group_records.get(group):
group_records[group] = len(h5_config[subgroup]) - 1
# index 0 is the metadata
print(f"Number of existing records: {group_records}")
print("Using subgroups of the config file and adding records from the h5 file.")
# check if config file subgroups exist in the h5 file
for subgroup in yaml_config:
value = [yaml_config[subgroup]] # first record is the metadata
if subgroup in h5_subgroups:
# h5_config[subgroup] is a list of dictionaries (records)
h5_subgroup_keys = list(h5_config[subgroup][0].keys())
for idx in range(1, nb_records + 1):
record = {}
for key in yaml_config[subgroup]:
for yaml_subgroup, inner_value in yaml_config.items():
# example of yaml_subgroup: '/Default/Sample'
group_name = "/" + yaml_subgroup.split(sep="/", maxsplit=2)[1]
value: List[Dict[str, str]] = [inner_value]
if h5_config.get(yaml_subgroup, None):
# h5_config[yaml_subgroup] is a list of dictionaries (records)
h5_subgroup_keys = list(h5_config[yaml_subgroup][0].keys())
for idx in range(1, group_records[group_name] + 1):
record: Dict[str, str] = {}
for key in inner_value:
if key in h5_subgroup_keys:
# h5_config[subgroup][idx] is a dict (a record)
record[key] = h5_config[subgroup][idx][key]
# h5_config[yaml_subgroup][idx] is a dict (a record)
record[key] = h5_config[yaml_subgroup][idx][key]
else:
record[key] = "None"
value.append(record)
else:
# fill nb_records with None, set the color to white
for idx in range(nb_records):
record = {}
for key in yaml_config[subgroup]:
record[key] = "#fff" if key == "Color" else "None"
value.append(record)
final_config[subgroup] = value
# -> for a new motor, None values have to be set for the number of records
# -> for a brand-new group, nothing to add
if group_records.get(group_name, None): # new motor for an existing group
for idx in range(group_records[group_name]):
record = {}
for key in inner_value:
record[key] = "#fff" if key == "Color" else "None"
value.append(record)
final_config[yaml_subgroup] = value
return final_config
......
import h5py
from pathlib import Path
import yaml
......@@ -58,6 +59,15 @@ new_config = {
},
}
inverted_config = {
"groups": ["Default", "Lab099"],
"motors": {
"Sample": ["X ['Position'][mm]", "Y ['Position'][mm]", "Z ['Position'][mm]"],
"Chiller": ["Ika ['Temperature'][degC]", "Ika ['Velocity'][rpm]"],
"Detector": ["Z ['Position'][mm]", "Y ['Position'][mm]", "X ['Position'][mm]"],
},
}
def test_get_existing_config_ok():
config = upd.get_existing_config(filename=h5file)
......@@ -151,4 +161,34 @@ def test_generate_config_ok():
assert "/Default/Chiller" in final_config
assert final_config["/Default/Detector"][1].get("Time") == "31-05-2021 12:05:59"
assert final_config["/Default/Chiller"][1].get("Time") == "None"
assert final_config["/Lab099/Detector"][1].get("Z ['Position'][mm]") == "None"
assert len(final_config["/Default/Detector"]) == 2
assert final_config["/Lab099/Detector"][0].get("Z ['Position'][mm]") == "None"
assert len(final_config["/Lab099/Detector"]) == 1
def test_generate_config_invert_keys():
final_config = upd.generate_config(
yaml_config=upd.check_config(inverted_config), h5_config=existing_config
)
assert "/Default/Chiller" in final_config
assert final_config["/Default/Detector"][1].get("Time") == "31-05-2021 12:05:59"
assert final_config["/Default/Chiller"][1].get("Time") == "None"
assert len(final_config["/Default/Detector"]) == 2
assert final_config["/Lab099/Detector"][0].get("Z ['Position'][mm]") == "None"
assert len(final_config["/Lab099/Detector"]) == 1
def test_generate_saved_attributes_inverted_config(tmp_path):
new_h5file = tmp_path / "saved_attributes_updated.h5"
final_config = upd.generate_config(
yaml_config=upd.check_config(inverted_config), h5_config=existing_config
)
assert upd.generate_saved_attributes(path=new_h5file, groups=final_config) is True
assert new_h5file.is_file()
saved_config = upd.get_existing_config(filename=new_h5file)
assert "/Lab099/Detector" in saved_config
assert saved_config["/Default/Sample"][1].get("Time") == "31-05-2021 12:05:59"
assert saved_config["/Default/Chiller"][1].get("Time") == "None"
assert len(saved_config["/Default/Detector"]) == 2
assert saved_config["/Lab099/Sample"][0].get("Z ['Position'][mm]") == "None"
assert len(saved_config["/Lab099/Chiller"]) == 1
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment