Skip to content
Snippets Groups Projects

Implement deadtime correction task

Merged Loic Huder requested to merge deadtime-corr into main
All threads resolved!
Files
5
+ 62
0
from typing import Any, Optional
import h5py
import numpy as np
import json
from . import nexus as nx
from ewoksfluo import __version__ as version
def create_entry_group(file: h5py.File, entry_name: str, default: Optional[str] = None):
entry_group = file.get(entry_name, None)
if entry_group is None:
entry_group = nx.create_entry(file, entry_name, default=default)
assert isinstance(entry_group, h5py.Group)
# Update default in case of an existing entry_group
entry_group.attrs["default"] = default
return entry_group
def create_process_group(
entry_group: h5py.Group,
process_name: str,
config: dict[str, Any],
default: Optional[str] = None,
):
process_group = nx.create_process(
entry_group,
process_name,
default=default,
)
process_group["program"] = "ewoksfluo"
process_group["version"] = version
config_group = process_group.create_group("configuration")
config_group.create_dataset("data", data=json.dumps(config))
config_group.create_dataset("date", data=nx.now())
config_group.create_dataset("type", data="json")
return process_group
def get_time_in_seconds(time_dset: h5py.Dataset) -> np.ndarray:
unit = time_dset.attrs.get("units", "s")
if unit == "ms":
return time_dset[()] / 1000
if unit == "s":
return time_dset[()]
raise ValueError(f"Unknown unit {unit} in {time_dset.name}")
def get_correction_factor(deadtime_uri: str, reference_time: float) -> np.ndarray:
deadtime_file, deadtime_h5path = deadtime_uri.split("::")
with h5py.File(deadtime_file, "r") as h5file:
deadtime_dset = h5file[deadtime_h5path]
assert isinstance(deadtime_dset, h5py.Dataset)
deadtime = get_time_in_seconds(deadtime_dset)
return deadtime / reference_time
Loading