190 lines
6.8 KiB
Python
190 lines
6.8 KiB
Python
import os
|
|
import nixio as nix
|
|
from enum import Enum
|
|
|
|
|
|
class ItemDescriptor():
|
|
|
|
def __init__(self, name=None, id=None, type=None, value=None, definition=None, block_id=None, entity_type=None) -> None:
|
|
super().__init__()
|
|
self.name = name
|
|
self.type = type
|
|
self.id = id
|
|
self.block_id= block_id
|
|
self.definition = definition
|
|
self.value = value
|
|
self.entity_type = entity_type
|
|
|
|
class NodeType(Enum):
|
|
Root = "root"
|
|
Section = "section"
|
|
Block = "block"
|
|
DataArray = "data_array"
|
|
DataFrame = "data_frame"
|
|
Property = "property"
|
|
Dimension = "dimension"
|
|
Source = "source"
|
|
Tag = "tag"
|
|
MultiTag = "multi_tag"
|
|
Group = "group"
|
|
Feature="feature"
|
|
|
|
|
|
class Singleton(type):
|
|
_instances = {}
|
|
def __call__(cls, *args, **kwargs):
|
|
if cls not in cls._instances:
|
|
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
|
|
return cls._instances[cls]
|
|
|
|
class FileHandler(metaclass=Singleton):
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self._filename = None
|
|
self._nix_file = None
|
|
self._file_requests = []
|
|
|
|
def open(self, filename):
|
|
"""[summary]
|
|
|
|
Args:
|
|
filename ([type]): [description]
|
|
"""
|
|
self.close()
|
|
|
|
if not os.path.exists(filename):
|
|
return False, "File %s could not be found!" % filename
|
|
try:
|
|
self._nix_file = nix.File.open(filename, nix.FileMode.ReadOnly)
|
|
self._filename = filename
|
|
return True, "Successfully opened file %s." % filename.split(os.sep)[-1]
|
|
except RuntimeError as e:
|
|
return False, "Failed to open file %s! \n Error message is: %s" % (filename, e)
|
|
except OSError as e:
|
|
return False, "Failed to open file %s! \n Error message is: %s\n Probably no nix file?!" % (filename, e)
|
|
|
|
def close(self):
|
|
# TODO check if there are any pending file requests!
|
|
if self._nix_file is not None and self._nix_file.is_open():
|
|
self._nix_file.close()
|
|
self._nix_file = None
|
|
self._file_requests = []
|
|
|
|
def file_descriptor(self):
|
|
|
|
return ItemDescriptor()
|
|
|
|
|
|
@property
|
|
def is_valid(self):
|
|
return self._nix_file is not None and self._nix_file.is_open()
|
|
|
|
@property
|
|
def filename(self):
|
|
return self._filename
|
|
|
|
def request_metadata(self, root_id=None, depth=1):
|
|
"""[summary]
|
|
|
|
Args:
|
|
root_id ([type], optional): [description]. Defaults to None.
|
|
depth (int, optional): [description]. Defaults to 1.
|
|
"""
|
|
def get_subsections(section):
|
|
sub_sections = []
|
|
for s in section.sections:
|
|
sub_sections.append(ItemDescriptor(s.name, s.id, s.type, definition=s.definition, entity_type=NodeType.Section))
|
|
return sub_sections
|
|
|
|
def get_properties(section):
|
|
props = []
|
|
for p in section.props:
|
|
value = "unset"
|
|
props.append(ItemDescriptor(p.name, p.id, value=value, entity_type=NodeType.Property))
|
|
return props
|
|
|
|
sections = []
|
|
properties = []
|
|
if root_id is None:
|
|
sections = get_subsections(self._nix_file)
|
|
else:
|
|
found_section = self._nix_file.find_sections(lambda s: s.id == root_id)
|
|
for fs in found_section:
|
|
sections.extend(get_subsections(fs))
|
|
properties.extend(get_properties(fs))
|
|
return sections, properties
|
|
|
|
def _entity_info(self, entities, block_id, entity_type):
|
|
infos = []
|
|
for e in entities:
|
|
itd = ItemDescriptor(e.name, e.id, e.type, definition=e.definition, entity_type=entity_type, block_id=block_id)
|
|
infos.append(itd)
|
|
# TODO set the value to something meaningful for the various entity types
|
|
return infos
|
|
|
|
def request_blocks(self):
|
|
return self._entity_info(self._nix_file.blocks, None, NodeType.Block)
|
|
|
|
def request_data_arrays(self, block_id):
|
|
return self._entity_info(self._nix_file.blocks[block_id].data_arrays, block_id, NodeType.DataArray)
|
|
|
|
def request_tags(self, block_id):
|
|
tags = self._entity_info(self._nix_file.blocks[block_id].tags, block_id, NodeType.Tag)
|
|
tags.extend(self._entity_info(self._nix_file.blocks[block_id].multi_tags, block_id, NodeType.MultiTag))
|
|
return tags
|
|
|
|
def request_references(self, block_id, tag_id, is_mtag):
|
|
b = self._nix_file.blocks[block_id]
|
|
t = None
|
|
if is_mtag:
|
|
t = b.multi_tags[tag_id]
|
|
else:
|
|
t = b.tags[tag_id]
|
|
return self._entity_info(t.references, block_id, NodeType.DataArray)
|
|
|
|
def request_features(self, block_id, tag_id, is_mtag):
|
|
b = self._nix_file.blocks[block_id]
|
|
t = None
|
|
if is_mtag:
|
|
t = b.multi_tags[tag_id]
|
|
else:
|
|
t = b.tags[tag_id]
|
|
feats = []
|
|
for f in t.features:
|
|
itd = ItemDescriptor(f.data.name, f.id, f.link_type, definition=f.data.definition, block_id=block_id, entity_type=NodeType.Feature)
|
|
feats.append(itd)
|
|
return feats
|
|
|
|
def request_dimensions(self, block_id, array_id):
|
|
dimensions = []
|
|
for i, d in enumerate(self._nix_file.blocks[block_id].data_arrays[array_id].dimensions):
|
|
dim_name = "%i. dim: %s" % (i+1, d.label if hasattr(d, "label") else "")
|
|
dim_type= "%s %s" % (d.dimension_type, "dimension")
|
|
dimensions.append(ItemDescriptor(dim_name, type=dim_type, entity_type=NodeType.Dimension, block_id=block_id))
|
|
return dimensions
|
|
|
|
def request_data_frames(self, block_id):
|
|
if self._nix_file.version[1] >= 2:
|
|
return self._entity_info(self._nix_file.blocks[block_id].data_frames, block_id, NodeType.DataFrame)
|
|
return []
|
|
|
|
def request_groups(self, block_id):
|
|
return self._entity_info(self._nix_file.blocks[block_id].groups, block_id, NodeType.Group)
|
|
|
|
def request_sources(self, block_id, parent_source_id=None):
|
|
def get_subsources(src):
|
|
sub_sources = []
|
|
for s in src.sources:
|
|
sub_sources.append(ItemDescriptor(s.name, s.id, s.type, definition=s.definition, entity_type=NodeType.Source))
|
|
return sub_sources
|
|
|
|
if parent_source_id is None:
|
|
return self._entity_info(self._nix_file.blocks[block_id].sources, block_id, NodeType.Source)
|
|
else:
|
|
b = self._nix_file.blocks[block_id]
|
|
srcs = b.find_sources(lambda s: s.id == parent_source_id)
|
|
sources = []
|
|
for src in srcs:
|
|
sources.extend(get_subsources(src))
|
|
return sources |