Skip to content
This repository was archived by the owner on Nov 6, 2025. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions src/pymodaq_data/h5modules/backends.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

@author: Sebastien Weber
"""
from typing import Union
from enum import Enum
import numpy as np
import importlib
from importlib import metadata
Expand Down Expand Up @@ -63,6 +65,7 @@ class SaveType(BaseEnum):
logger = 2
custom = 3
actuator = 4
optimizer = 5


class GroupType(BaseEnum):
Expand Down Expand Up @@ -1008,14 +1011,14 @@ def create_vlarray(self, where, name, dtype, title=''):
array.attrs['backend'] = self.backend
return array

def add_group(self, group_name, group_type: GroupType, where, title='', metadata=dict([])) -> GROUP:
def add_group(self, group_name, group_type: Union[GroupType, str], where, title='', metadata=dict([])) -> GROUP:
"""
Add a node in the h5 file tree of the group type
Parameters
----------
group_name: (str) a custom name for this group
group_type: str or GroupType enum
one of the possible values of GroupType
one of the possible values of GroupType, should be enforced by higher level modules not here
where: (str or node) parent node where to create the new group
metadata: (dict) extra metadata to be saved with this new group node

Expand All @@ -1026,14 +1029,15 @@ def add_group(self, group_name, group_type: GroupType, where, title='', metadata
if isinstance(where, Node):
where = where.node

group_type = enum_checker(GroupType, group_type)
if isinstance(group_type, Enum):
group_type = group_type.name

if group_name in self.get_children(self.get_node(where)):
node = self.get_node(where, group_name)

else:
node = self.get_set_group(where, utils.capitalize(group_name), title)
node.attrs['type'] = group_type.name.lower()
node.attrs['type'] = group_type.lower()
for metadat in metadata:
node.attrs[metadat] = metadata[metadat]
node.attrs['backend'] = self.backend
Expand Down
65 changes: 48 additions & 17 deletions src/pymodaq_data/h5modules/saving.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
"""
import copy
import datetime
import enum

from dateutil import parser
from numbers import Number
import os
Expand Down Expand Up @@ -262,16 +264,18 @@ def get_set_group(self, where, name, title='', **kwargs):
self._current_group = super().get_set_group(where, name, title, **kwargs)
return self._current_group

def get_groups(self, where: Union[str, GROUP], group_type: GroupType):
def get_groups(self, where: Union[str, GROUP], group_type: Union[str, GroupType, BaseEnum]):
"""Get all groups hanging from a Group and of a certain type"""
groups = []
if isinstance(group_type, enum.Enum):
group_type = group_type.name
for node_name in list(self.get_children(where)):
group = self.get_node(where, node_name)
if 'type' in group.attrs and group.attrs['type'] == group_type.name:
if 'type' in group.attrs and group.attrs['type'].lower() == group_type.lower():
groups.append(group)
return groups

def get_last_group(self, where: GROUP, group_type: GroupType):
def get_last_group(self, where: GROUP, group_type: Union[str, GroupType, enum.Enum]):
groups = self.get_groups(where, group_type)
if len(groups) != 0:
return groups[-1]
Expand All @@ -288,7 +292,7 @@ def get_node_from_title(self, where, title: str):
"""Get a Node starting from a given node (Group) matching the given title"""
return self.get_node_from_attribute_match(where, 'TITLE', title)

def add_data_group(self, where, data_dim: DataDim, title='', settings_as_xml='', metadata=dict([])):
def add_data_group(self, where, data_dim: DataDim, title='', settings_as_xml='', metadata=None):
"""Creates a group node at given location in the tree

Parameters
Expand All @@ -311,12 +315,14 @@ def add_data_group(self, where, data_dim: DataDim, title='', settings_as_xml='',
--------
:py:meth:`add_group`
"""
if metadata is None:
metadata = {}
data_dim = enum_checker(DataDim, data_dim)
metadata.update(settings=settings_as_xml)
group = self.add_group(data_dim.name, 'data_dim', where, title, metadata)
return group

def add_incremental_group(self, group_type, where, title='', settings_as_xml='', metadata=dict([])):
def add_incremental_group(self, group_type: Union[str, GroupType, enum.Enum], where, title='', settings_as_xml='', metadata=None):
"""
Add a node in the h5 file tree of the group type with an increment in the given name
Parameters
Expand All @@ -336,79 +342,104 @@ def add_incremental_group(self, group_type, where, title='', settings_as_xml='',
-------
node: newly created group node
"""
group_type = enum_checker(GroupType, group_type)
if metadata is None:
metadata = {}
if isinstance(group_type, enum.Enum):
group_type = group_type.name

nodes = [name for name in self.get_children(self.get_node(where))]
nodes_tmp = []
for node in nodes:
if utils.capitalize(group_type.name) in node:
if utils.capitalize(group_type.lower()) in node:
nodes_tmp.append(node)
nodes_tmp.sort()
if len(nodes_tmp) == 0:
ind_group = -1
else:
ind_group = int(nodes_tmp[-1][-3:])
group = self.get_set_group(where, f'{utils.capitalize(group_type.name)}{ind_group + 1:03d}', title)
group = self.get_set_group(where, f'{utils.capitalize(group_type.lower())}{ind_group + 1:03d}', title)
self.set_attr(group, 'settings', settings_as_xml)
if group_type.name.lower() != 'ch':
self.set_attr(group, 'type', group_type.name.lower())
if group_type.lower() != 'ch':
self.set_attr(group, 'type', group_type.lower())
else:
self.set_attr(group, 'type', '')
for metadat in metadata:
self.set_attr(group, metadat, metadata[metadat])
return group

def add_act_group(self, where, title='', settings_as_xml='', metadata=dict([])):
def add_act_group(self, where, title='', settings_as_xml='', metadata=None):
"""
Add a new group of type detector
See Also
-------
add_incremental_group
"""
if metadata is None:
metadata = {}
group = self.add_incremental_group('actuator', where, title, settings_as_xml, metadata)
return group

def add_det_group(self, where, title='', settings_as_xml='', metadata=dict([])):
def add_det_group(self, where, title='', settings_as_xml='', metadata=None):
"""
Add a new group of type detector
See Also
-------
add_incremental_group
"""
if metadata is None:
metadata = {}
group = self.add_incremental_group('detector', where, title, settings_as_xml, metadata)
return group

def add_scan_group(self, where='/RawData', title='', settings_as_xml='', metadata=dict([])):
"""Add a new group of type scan
def add_generic_group(self, where='/RawData', title='', settings_as_xml='', metadata=None,
group_type=GroupType.scan):
"""Add a new group of type given by the input argument group_type

At creation adds the attributes description and scan_done to be used elsewhere

See Also
-------
add_incremental_group
"""
if metadata is None:
metadata = {}
metadata.update(dict(description='', scan_done=False))
group = self.add_incremental_group(group_type, where, title, settings_as_xml, metadata)
return group

def add_scan_group(self, where='/RawData', title='', settings_as_xml='', metadata=None,):
"""Add a new group of type scan

deprecated, use add_generic_group with a group type as GroupType.scan
"""
if metadata is None:
metadata = {}
metadata.update(dict(description='', scan_done=False))
group = self.add_incremental_group(GroupType['scan'], where, title, settings_as_xml, metadata)
group = self.add_generic_group(where, title, settings_as_xml, metadata, group_type=GroupType.scan)
return group

def add_ch_group(self, where, title='', settings_as_xml='', metadata=dict([])):
def add_ch_group(self, where, title='', settings_as_xml='', metadata=None):
"""
Add a new group of type channel
See Also
-------
add_incremental_group
"""
if metadata is None:
metadata = {}
group = self.add_incremental_group('ch', where, title, settings_as_xml, metadata)
return group


def add_move_group(self, where, title='', settings_as_xml='', metadata=dict([])):
def add_move_group(self, where, title='', settings_as_xml='', metadata=None):
"""
Add a new group of type actuator
See Also
-------
add_incremental_group
"""
if metadata is None:
metadata = {}
group = self.add_incremental_group('actuator', where, title, settings_as_xml, metadata)
return group

Expand Down
7 changes: 4 additions & 3 deletions tests/h5module_test/backend_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,9 +154,10 @@ def test_add_group(self, get_backend, group_type):
assert g3.attrs['type'] == group_type
assert g3.attrs['attr1'] == 'attr1'
assert g3.attrs['attr2'] == 21.4
gtype = 'this is not a valid group type'
with pytest.raises(ValueError):
g4 = bck.add_group('g4', gtype, bck.root())
# this below is not enforced anymore pymodaq_data>5.0.20
# gtype = 'this is not a valid group type'
# with pytest.raises(ValueError):
# g4 = bck.add_group('g4', gtype, bck.root())
bck.close_file()

def test_group_creation(self, get_backend):
Expand Down