Source code for malcolm.modules.scanning.parts.datasettablepart

from typing import List, Optional

from annotypes import add_call_types

from malcolm.core import AttributeModel, Part, PartRegistrar, TableMeta
from malcolm.modules import builtin

from ..hooks import APartInfo, PostConfigureHook
from ..infos import DatasetProducedInfo
from ..util import DatasetTable


[docs]class DatasetTablePart(Part): """Exposes an Attribute that reports the datasets that will be written during a scan""" datasets: Optional[AttributeModel] = None def setup(self, registrar: PartRegistrar) -> None: self.datasets = TableMeta.from_table( DatasetTable, "Datasets produced in HDF file" ).create_attribute_model() registrar.add_attribute_model("datasets", self.datasets) # Hooks registrar.hook(PostConfigureHook, self.on_post_configure) registrar.hook(builtin.hooks.ResetHook, self.on_reset)
[docs] @add_call_types def on_post_configure(self, part_info: APartInfo) -> None: # Update the dataset table name, filename, typ, rank, path, uid = [], [], [], [], [], [] i: DatasetProducedInfo filtered_values: List[DatasetProducedInfo] = DatasetProducedInfo.filter_values( part_info ) for i in filtered_values: name.append(i.name) filename.append(i.filename) typ.append(i.type) rank.append(i.rank) path.append(i.path) uid.append(i.uniqueid) datasets_table = DatasetTable(name, filename, typ, rank, path, uid) assert self.datasets, "No datasets" self.datasets.set_value(datasets_table)
def on_reset(self): self.datasets.set_value(None)