Source code for cdl.core.gui.h5io
# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.
"""
HDF5 I/O
========
The :mod:`cdl.core.gui.h5io` module provides the HDF5 file open/save into/from
DataLab data model/main window.
.. autoclass:: H5InputOutput
"""
from __future__ import annotations
import os.path as osp
from typing import TYPE_CHECKING
from guidata.qthelpers import exec_dialog
from qtpy import QtWidgets as QW
from cdl.config import _
from cdl.core.io.h5 import H5Importer
from cdl.core.io.native import NativeH5Reader, NativeH5Writer
from cdl.core.model.signal import SignalObj
from cdl.env import execenv
from cdl.utils.qthelpers import create_progress_bar, qt_try_loadsave_file
from cdl.widgets.h5browser import H5BrowserDialog
if TYPE_CHECKING:
from cdl.core.gui.main import CDLMainWindow
from cdl.core.io.h5.common import BaseNode
[docs]
class H5InputOutput:
"""Object handling HDF5 file open/save into/from DataLab data model/main window
Args:
mainwindow: Main window
"""
def __init__(self, mainwindow: CDLMainWindow) -> None:
self.mainwindow = mainwindow
self.uint32_wng: bool = None
@staticmethod
def __progbartitle(fname: str) -> str:
"""Return progress bar title"""
return _("Loading data from %s...") % osp.basename(fname)
[docs]
def save_file(self, filename: str) -> None:
"""Save all signals and images from DataLab model into a HDF5 file"""
writer = NativeH5Writer(filename)
for panel in self.mainwindow.panels:
panel.serialize_to_hdf5(writer)
writer.close()
[docs]
def open_file(self, filename: str, import_all: bool, reset_all: bool) -> None:
"""Open HDF5 file"""
progress = None
try:
reader = NativeH5Reader(filename)
if reset_all:
self.mainwindow.reset_all()
with create_progress_bar(
self.mainwindow, self.__progbartitle(filename), 2
) as progress:
for idx, panel in enumerate(self.mainwindow.panels):
progress.setValue(idx + 1)
QW.QApplication.processEvents()
panel.deserialize_from_hdf5(reader)
if progress.wasCanceled():
break
reader.close()
except KeyError:
if progress is not None:
# KeyError was encoutered when deserializing datasets (DataLab data
# model is not compatible with this version)
progress.close()
self.import_files([filename], import_all, reset_all)
def __add_object_from_node(self, node: BaseNode) -> None:
"""Add DataLab object from h5 node"""
obj = node.get_native_object()
if obj is None:
return
self.uint32_wng = self.uint32_wng or node.uint32_wng
if isinstance(obj, SignalObj):
self.mainwindow.signalpanel.add_object(obj)
else:
self.mainwindow.imagepanel.add_object(obj)
def __eventually_show_warnings(self) -> None:
"""Eventually show warnings after everything is imported"""
if self.uint32_wng:
QW.QMessageBox.warning(
self.mainwindow, _("Warning"), _("Clipping uint32 data to int32.")
)
[docs]
def import_files(
self, filenames: list[str], import_all: bool, reset_all: bool
) -> None:
"""Import HDF5 files"""
h5browser = H5BrowserDialog(self.mainwindow)
for filename in filenames:
with qt_try_loadsave_file(self.mainwindow, filename, "load"):
h5browser.open_file(filename)
if h5browser.is_empty():
h5browser.cleanup()
if not execenv.unattended:
QW.QMessageBox.warning(
self.mainwindow,
_("Warning"),
_("No supported data available in HDF5 file(s)."),
)
return
if execenv.unattended:
# Unattended mode: import all datasets (for testing)
import_all = True
if import_all or exec_dialog(h5browser):
if import_all:
nodes = h5browser.get_all_nodes()
else:
nodes = h5browser.get_nodes()
if nodes is not None:
if reset_all:
self.mainwindow.reset_all()
with qt_try_loadsave_file(self.mainwindow, "*.h5", "load"):
with create_progress_bar(self.mainwindow, "", len(nodes)) as prog:
self.uint32_wng = False
for idx, node in enumerate(nodes):
prog.setLabelText(self.__progbartitle(node.h5file.filename))
prog.setValue(idx + 1)
QW.QApplication.processEvents()
if prog.wasCanceled():
break
self.__add_object_from_node(node)
self.__eventually_show_warnings()
h5browser.cleanup()
[docs]
def import_dataset_from_file(self, filename: str, dsetname: str) -> None:
"""Import dataset from HDF5 file"""
h5importer = H5Importer(filename)
try:
node = h5importer.get(dsetname)
self.uint32_wng = False
self.__add_object_from_node(node)
self.__eventually_show_warnings()
except KeyError as exc:
raise KeyError(f"Dataset not found: {dsetname}") from exc
h5importer.close()