# -*- coding: utf-8 -*- # Copyright © 2014 - 2022, German Neuroinformatics Node (G-Node) # # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted under the terms of the BSD License. See # LICENSE file in the root of the Project. import warnings import numpy as np from .entity import Entity from .source_link_container import SourceLinkContainer from .container import Container, LinkContainer from .datatype import DataType from .data_array import DataArray from .data_view import DataView from .feature import Feature from .exceptions import (OutOfBounds, IncompatibleDimensions, UninitializedEntity, InvalidUnit) from .dimension_type import DimensionType from .link_type import LinkType from . import util from .section import Section from .dimensions import SliceMode class FeatureContainer(Container): """ The FeatureContainer has one minor difference from the regular Container: A Feature can be retrieved by the ID or name of the linked DataArray as well as the ID of the feature itself. """ def __getitem__(self, item): try: return Container.__getitem__(self, item) except KeyError as exc: # item might be the ID of the referenced data; try it as well for feat in self: if feat.data.id == item or feat.data.name == item: return feat raise exc def __contains__(self, item): if isinstance(item, Feature): item = item.id if not Container.__contains__(self, item): # check if it contains a Feature whose data matches 'item' for feat in self: if feat.data.id == item or feat.data.name == item: return True return False return True class BaseTag(Entity): """ Base class for Tag and MultiTag """ def __init__(self, nixfile, nixparent, h5group): super(BaseTag, self).__init__(nixfile, nixparent, h5group) self._sources = None self._references = None self._features = None @property def units(self): """ Property containing the units of the tag. The tag must provide a unit for each dimension of the position or extent vector. This is a read-write property. :type: list of str """ return tuple(u.decode() if isinstance(u, bytes) else u for u in self._h5group.get_data("units")) @units.setter def units(self, units): if not units: if self._h5group.has_data("units"): del self._h5group["units"] else: sanitized = [] for unit in units: util.check_attr_type(unit, str) unit = util.units.sanitizer(unit) sanitized.append(unit) dtype = DataType.String self._h5group.write_data("units", sanitized, dtype) if self.file.auto_update_timestamps: self.force_updated_at() def create_feature(self, data, link_type): """ Create a new feature. :param data: The data array of this feature. :type data: nixio.DataArray :param link_type: The link type of this feature. :type link_type: nixio.LinkType :returns: The created feature object. :rtype: nixio.Feature """ if isinstance(link_type, str): link_type = link_type.lower() link_type = LinkType(link_type) features = self._h5group.open_group("features") feat = Feature.create_new(self.file, self, features, data, link_type) return feat def _calc_data_slices(self, data, position, extent, stop_rule): refslice = list() if not self.units: units = [None] * len(data.dimensions) else: units = self.units for idx, dim in enumerate(data.dimensions): if idx < len(position): start_pos = position[idx] start_pos, scaling = self._scale_position(start_pos, units[idx], dim) if extent is not None and idx < len(extent): stop_pos = extent[idx] stop_pos *= scaling stop_pos += start_pos slice_mode = stop_rule if extent[idx] > 0.0 else SliceMode.Inclusive else: stop_pos = start_pos slice_mode = SliceMode.Inclusive range_indices = dim.range_indices(start_pos, stop_pos, slice_mode) refslice.append(range_indices if range_indices is None else slice(range_indices[0], range_indices[1] + 1)) else: # no position, we take the whole slice for this dimension start_index = 0 stop_index = data.shape[idx] refslice.append(slice(start_index, stop_index)) return tuple(refslice) @staticmethod def _slices_in_data(data, slices): if slices is None or not all(slices): return False dasize = data.data_extent stops = tuple(sl.stop for sl in slices) return np.all(np.less_equal(stops, dasize)) @staticmethod def _scale_position(pos, unit, dim): dimtype = dim.dimension_type if dimtype == DimensionType.Set: dimunit = None else: dimunit = dim.unit scaling = 1.0 if dimtype == DimensionType.Set: if unit and unit != "none": raise IncompatibleDimensions( "Cannot apply a position with unit to a SetDimension", "Tag._pos_to_idx" ) else: if dimunit is None and unit is not None: raise IncompatibleDimensions( "If a unit if given for the position the dimension must not be without a unit!", "Tag._pos_to_idx" ) elif dimunit is not None and unit is not None: try: scaling = util.units.scaling(unit, dimunit) except InvalidUnit: raise IncompatibleDimensions( "Cannot scale Tag unit {} to match dimension unit {}".format(unit, dimunit), "Tag._pos_to_idx" ) return pos * scaling, scaling @staticmethod def _pos_to_idx(pos, unit, dim, mode): dimtype = dim.dimension_type if dimtype == DimensionType.Set: dimunit = None else: dimunit = dim.unit scaling = 1.0 if dimtype == DimensionType.Sample: if not dimunit and unit is not None: raise IncompatibleDimensions( "Units of position and SampledDimension " "must both be given!", "Tag._pos_to_idx" ) if dimunit and unit is not None: try: scaling = util.units.scaling(unit, dimunit) except InvalidUnit: raise IncompatibleDimensions( "Cannot scale Tag unit {} to match dimension unit {}".format(unit, dimunit), "Tag._pos_to_idx" ) index = dim.index_of(pos * scaling, mode) elif dimtype == DimensionType.Set: if unit and unit != "none": raise IncompatibleDimensions( "Cannot apply a position with unit to a SetDimension", "Tag._pos_to_idx" ) index = dim.index_of(pos, mode) else: # dimtype == DimensionType.Range: if dimunit and unit is not None: try: scaling = util.units.scaling(unit, dimunit) except InvalidUnit: raise IncompatibleDimensions( "Cannot scale Tag unit {} to match dimension unit {}".format(unit, dimunit), "Tag._pos_to_idx" ) index = dim.index_of(pos * scaling, mode) return int(index) @property def features(self): """ A property containing all features. Features can be obtained via their index or their ID. Features can be deleted from the list. Adding new features is done using the create_feature method. This is a read only attribute. :type: Container of nixio.Feature. """ if self._features is None: self._features = FeatureContainer("features", self.file, self, Feature) return self._features class Tag(BaseTag): @classmethod def create_new(cls, nixfile, nixparent, h5parent, name, type_, position): newentity = super(Tag, cls).create_new(nixfile, nixparent, h5parent, name, type_) newentity.position = position return newentity @property def position(self): """ The position defined by the tag. This is a read-write property. :type: list of float """ return tuple(self._h5group.get_data("position")) @position.setter def position(self, pos): if pos is not None and not hasattr(pos, "__getitem__"): pos = [pos] if pos is None or len(pos) == 0: if self._h5group.has_data("position"): del self._h5group["position"] else: dtype = DataType.Double self._h5group.write_data("position", pos, dtype) if self.file.auto_update_timestamps: self.force_updated_at() @property def extent(self): """ The extent defined by the tag. This is an optional read-write property and may be set to None. :type: list of float """ return tuple(self._h5group.get_data("extent")) @extent.setter def extent(self, ext): if ext is not None and not hasattr(ext, "__getitem__"): ext = [ext] if ext is None or len(ext) == 0: if self._h5group.has_data("extent"): del self._h5group["extent"] else: dtype = DataType.Double self._h5group.write_data("extent", ext, dtype) if self.file.auto_update_timestamps: self.force_updated_at() def retrieve_data(self, refidx): msg = ("Call to deprecated method Tag.retrieve_data. " "Use Tag.tagged_data instead.") warnings.warn(msg, category=DeprecationWarning) return self.tagged_data(refidx) def tagged_data(self, refidx, stop_rule=SliceMode.Exclusive): references = self.references position = self.position extent = self.extent if len(references) == 0: raise OutOfBounds("There are no references in this tag!") if isinstance(refidx, int) and refidx >= len(references): raise OutOfBounds("Reference index out of bounds.") ref = references[refidx] if extent and len(position) != len(extent): raise IncompatibleDimensions( "Number of dimensions in position and extent " "do not match ", extent) slices = self._calc_data_slices(ref, self.position, self.extent, stop_rule) if not all(slices): return DataView(ref, slices) if not self._slices_in_data(ref, slices): raise OutOfBounds("References data slice out of the extent of the " "DataArray!") return DataView(ref, slices) def retrieve_feature_data(self, featidx): msg = ("Call to deprecated method Tag.retrieve_feature_data. " "Use Tag.feature_data instead.") warnings.warn(msg, category=DeprecationWarning) return self.feature_data(featidx) def feature_data(self, featidx, stop_rule=SliceMode.Exclusive): if len(self.features) == 0: raise OutOfBounds("There are no features associated with this tag!") try: feat = self.features[featidx] except KeyError: feat = None for feature in self.features: if feature.data.name == featidx or feature.data.id == featidx: feat = feature break if feat is None: raise data = feat.data if data is None: raise UninitializedEntity() if feat.link_type == LinkType.Tagged: slices = self._calc_data_slices(data, self.position, self.extent, stop_rule) if not self._slices_in_data(data, slices): raise OutOfBounds("Requested data slice out of the extent " "of the Feature!") return DataView(data, slices) # For untagged and indexed return the full data fullslices = tuple(slice(0, stop) for stop in data.shape) return DataView(data, fullslices) @property def references(self): """ A property containing all data arrays referenced by the tag. Referenced data arrays can be obtained by index or their id. References can be removed from the list, removing a referenced DataArray will not remove it from the file. New references can be added using the append method of the list. This is a read only attribute. :type: Container of nixio.DataArray """ if self._references is None: self._references = LinkContainer("references", self, DataArray, self._parent.data_arrays) return self._references @property def sources(self): """ A property containing all Sources referenced by the Tag. Sources can be obtained by index or their id. Sources can be removed from the list, but removing a referenced Source will not remove it from the file. New Sources can be added using the append method of the list. This is a read only attribute. """ if self._sources is None: self._sources = SourceLinkContainer(self) return self._sources # metadata @property def metadata(self): """ Associated metadata of the entity. Sections attached to the entity via this attribute can provide additional annotations. This is an optional read-write property, and can be None if no metadata is available. :type: nixio.Section """ if "metadata" in self._h5group: return Section(self.file, None, self._h5group.open_group("metadata")) return None @metadata.setter def metadata(self, sect): if not isinstance(sect, Section): raise TypeError("{} is not of type Section".format(sect)) self._h5group.create_link(sect, "metadata") @metadata.deleter def metadata(self): if "metadata" in self._h5group: self._h5group.delete("metadata")