Source code for geoh5py.shared.entity

#  Copyright (c) 2024 Mira Geoscience Ltd.
#  This file is part of geoh5py.
#  geoh5py is free software: you can redistribute it and/or modify
#  it under the terms of the GNU Lesser General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#  geoh5py is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  GNU Lesser General Public License for more details.
#  You should have received a copy of the GNU Lesser General Public License
#  along with geoh5py.  If not, see <>.

# pylint: disable=R0904

from __future__ import annotations

import uuid
import warnings
from abc import ABC, abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING

import numpy as np

from geoh5py.shared.utils import str2uuid

    from numpy import ndarray

    from .. import shared
    from ..groups import PropertyGroup
    from ..workspace import Workspace

DEFAULT_CRS = {"Code": "Unknown", "Name": "Unknown"}

[docs] class Entity(ABC): """ Base Entity class """ _attribute_map: dict = { "Allow delete": "allow_delete", "Allow move": "allow_move", "Allow rename": "allow_rename", "Clipping IDs": "clipping_ids: list | None", "ID": "uid", "Name": "name", "Partially hidden": "partially_hidden", "Public": "public", "Visible": "visible", } _visible = True def __init__(self, uid: uuid.UUID | None = None, name="Entity", **kwargs): self._uid = ( str2uuid(uid) if isinstance(str2uuid(uid), uuid.UUID) else uuid.uuid4() ) self._name = name self._parent: Entity | None = None self._children: list = [] self._allow_delete = True self._allow_move = True self._allow_rename = True self._partially_hidden = False self._clipping_ids: list[uuid.UUID] | None = None self._public = True self._on_file = False self._metadata: dict | None = None for attr, item in kwargs.items(): try: if attr in self._attribute_map: attr = self._attribute_map[attr] setattr(self, attr, item) except AttributeError: continue
[docs] def add_children(self, children: list[shared.Entity]): """ :param children: Add a list of entities as :obj:`~geoh5py.shared.entity.Entity.children` """ for child in children: if child not in self._children: self._children.append(child)
[docs] def add_file(self, file: str): """ Add a file to the object or group stored as bytes on a FilenameData :param file: File name with path to import. """ if not Path(file).is_file(): raise ValueError(f"Input file '{file}' does not exist.") with open(file, "rb") as raw_binary: blob = name = Path(file).name attributes = { "name": name, "file_name": name, "association": "OBJECT", "parent": self, "values": blob, } entity_type = {"name": "UserFiles", "primitive_type": "FILENAME"} file_data = self.workspace.create_entity( None, entity=attributes, entity_type=entity_type ) return file_data
@property def allow_delete(self) -> bool: """ :obj:`bool` Entity can be deleted from the workspace. """ return self._allow_delete @allow_delete.setter def allow_delete(self, value: bool): self._allow_delete = value self.workspace.update_attribute(self, "attributes") @property def allow_move(self) -> bool: """ :obj:`bool` Entity can change :obj:`~geoh5py.shared.entity.Entity.parent` """ return self._allow_move @allow_move.setter def allow_move(self, value: bool): self._allow_move = value self.workspace.update_attribute(self, "attributes") @property def allow_rename(self) -> bool: """ :obj:`bool` Entity can change name """ return self._allow_rename @allow_rename.setter def allow_rename(self, value: bool): self._allow_rename = value self.workspace.update_attribute(self, "attributes") @property def attribute_map(self) -> dict: """ :obj:`dict` Correspondence map between property names used in geoh5py and geoh5. """ return self._attribute_map @property def children(self): """ :obj:`list` Children entities in the workspace tree """ return self._children @property def clipping_ids(self) -> list[uuid.UUID] | None: """ List of clipping uuids """ return self._clipping_ids
[docs] @abstractmethod def mask_by_extent( self, extent: np.ndarray, inverse: bool = False ) -> np.ndarray | None: """ Get a mask array from coordinate extent. :param extent: Bounding box extent coordinates defined by either: - obj:`numpy.ndarray` of shape (2, 3) 3D coordinate: [[west, south, bottom], [east, north, top]] - obj:`numpy.ndarray` of shape (2, 2) Horizontal coordinates: [[west, south], [east, north]]. :param inverse: Return the complement of the mask extent. Default to False :return: Array of bool defining the vertices or cell centers within the mask extent, or None if no intersection. """
[docs] @classmethod def create(cls, workspace, **kwargs): """ Function to create an entity. :param workspace: Workspace to be added to. :param kwargs: List of keyword arguments defining the properties of a class. :return entity: Registered Entity to the workspace. """ entity_type_kwargs = ( {"entity_type": {"uid": kwargs["entity_type_uid"]}} if "entity_type_uid" in kwargs else {} ) entity_kwargs = {"entity": kwargs} new_object = workspace.create_entity( cls, **{**entity_kwargs, **entity_type_kwargs}, ) return new_object
@property def coordinate_reference_system(self) -> dict: """ Coordinate reference system attached to the entity. """ coordinate_reference_system = DEFAULT_CRS if self.metadata is not None and "Coordinate Reference System" in self.metadata: coordinate_reference_system = self.metadata[ "Coordinate Reference System" ].get("Current", DEFAULT_CRS) return coordinate_reference_system @coordinate_reference_system.setter def coordinate_reference_system(self, value: dict): # assert value is a dictionary containing "Code" and "Name" keys if not isinstance(value, dict): raise TypeError("Input coordinate reference system must be a dictionary") if value.keys() != {"Code", "Name"}: raise KeyError( "Input coordinate reference system must only contain a 'Code' and 'Name' keys" ) # get the actual coordinate reference system coordinate_reference_system = { "Current": value, "Previous": self.coordinate_reference_system, } # update the metadata metadata = self.metadata if isinstance(metadata, dict): metadata["Coordinate Reference System"] = coordinate_reference_system else: metadata = {"Coordinate Reference System": coordinate_reference_system} self.metadata = metadata
[docs] @abstractmethod def copy( self, parent=None, copy_children: bool = True, clear_cache: bool = False, mask: np.ndarray | None = None, **kwargs, ): """ Function to copy an entity to a different parent entity. :param parent: Target parent to copy the entity under. Copied to current :obj:`~geoh5py.shared.entity.Entity.parent` if None. :param copy_children: (Optional) Create copies of all children entities along with it. :param clear_cache: Clear array attributes after copy to minimize the memory footprint of the workspace. :param mask: Array of indices to sub-sample the input entity. :param kwargs: Additional keyword arguments to pass to the copy constructor. :return entity: Registered Entity to the workspace. """
[docs] def copy_from_extent( self, extent: ndarray, parent=None, copy_children: bool = True, clear_cache: bool = False, inverse: bool = False, **kwargs, ) -> Entity | None: """ Function to copy an entity to a different parent entity. :param extent: Bounding box extent requested for the input entity, as supplied for :func:`~geoh5py.shared.entity.Entity.mask_by_extent`. :param parent: Target parent to copy the entity under. Copied to current :obj:`~geoh5py.shared.entity.Entity.parent` if None. :param copy_children: (Optional) Create copies of all children entities along with it. :param clear_cache: Clear array attributes after copy. :param inverse: Keep the inverse (clip) of the extent selection. :param kwargs: Additional keyword arguments to pass to the copy constructor. :return entity: Registered Entity to the workspace. """ indices = self.mask_by_extent(extent, inverse=inverse) if indices is None: return None return self.copy( parent=parent, copy_children=copy_children, clear_cache=clear_cache, mask=indices, **kwargs, )
@property @abstractmethod def entity_type(self) -> shared.EntityType: ...
[docs] @classmethod def fix_up_name(cls, name: str) -> str: """If the given name is not a valid one, transforms it to make it valid :return: a valid name built from the given name. It simply returns the given name if it was already valid. """ # TODO: implement an actual fixup # (possibly it has to be abstract with different implementations per Entity type) return name
[docs] def get_entity(self, name: str | uuid.UUID) -> list[Entity | None]: """ Get a child :obj:`` by name. :param name: Name of the target child data :param entity_type: Sub-select entities based on type. :return: A list of children Data objects """ if isinstance(name, uuid.UUID): entity_list = [child for child in self.children if child.uid == name] else: entity_list = [child for child in self.children if == name] if not entity_list: return [None] return entity_list
[docs] def get_entity_list(self, entity_type=ABC) -> list[str]: """ Get a list of names of all children :obj:``. :param entity_type: Option to sub-select based on type. :return: List of names of data associated with the object. """ name_list = [ for child in self.children if isinstance(child, entity_type) ] return sorted(name_list)
@property def metadata(self) -> dict | None: """ Metadata attached to the entity. """ if getattr(self, "_metadata", None) is None: self._metadata = self.workspace.fetch_metadata(self.uid) return self._metadata @metadata.setter def metadata(self, value: dict | None): if value is not None: assert isinstance( value, (dict, str) ), f"Input metadata must be of type {dict} or None" self._metadata = value self.workspace.update_attribute(self, "metadata") @property def name(self) -> str: """ :obj:`str` Name of the entity """ return self._name @name.setter def name(self, new_name: str): self._name = self.fix_up_name(new_name) self.workspace.update_attribute(self, "attributes") @property def on_file(self) -> bool: """ Whether this Entity is already stored on :obj:`~geoh5py.workspace.workspace.Workspace.h5file`. """ return self._on_file @on_file.setter def on_file(self, value: bool): self._on_file = value @property def parent(self): return self._parent @parent.setter def parent(self, parent: shared.Entity): current_parent = self._parent if parent is not None: parent.add_children([self]) self._parent = parent if current_parent is not None and current_parent != self._parent: current_parent.remove_children([self]) self.workspace.save_entity(self) @property def partially_hidden(self) -> bool: """ Whether this Entity is partially hidden. """ return self._partially_hidden @partially_hidden.setter def partially_hidden(self, value: bool): self._partially_hidden = value self.workspace.update_attribute(self, "attributes") @property def public(self) -> bool: """ Whether this Entity is accessible in the workspace tree and other parts of the the user interface in ANALYST. """ return self._public @public.setter def public(self, value: bool): self._public = value self.workspace.update_attribute(self, "attributes")
[docs] def reference_to_uid( self, value: Entity | PropertyGroup | str | uuid.UUID ) -> list[uuid.UUID]: """ General entity reference translation. :param value: Either an `Entity`, string or uuid :return: List of unique identifier associated with the input reference. """ children_uid = [child.uid for child in self.children] if hasattr(value, "uid"): uid = [value.uid] elif isinstance(value, str): uid = [ obj.uid for obj in self.workspace.get_entity(value) if (obj is not None) and (obj.uid in children_uid) ] elif isinstance(value, uuid.UUID): uid = [value] return uid
[docs] def remove_children(self, children: list[shared.Entity] | list[PropertyGroup]): """ Remove children from the list of children entities. :param children: List of entities .. warning:: Removing a child entity without re-assigning it to a different parent may cause it to become inactive. Inactive entities are removed from the workspace by :func:`~geoh5py.shared.weakref_utils.remove_none_referents`. """ if not isinstance(children, list): children = [children] self._children = [child for child in self._children if child not in children] self.workspace.remove_children(self, children)
[docs] def save(self, add_children: bool = True): """ Alias method of :func:`~geoh5py.workspace.Workspace.save_entity`. WILL BE DEPRECATED AS ENTITIES ARE ALWAYS AUTOMATICALLY UPDATED. :param add_children: Option to also save the children. """ warnings.warn( " is deprecated and will be removed in next versions.", DeprecationWarning, ) return self.workspace.save_entity(self, add_children=add_children)
@property def uid(self) -> uuid.UUID: return self._uid @uid.setter def uid(self, uid: str | uuid.UUID): if isinstance(uid, str): uid = uuid.UUID(uid) self._uid = uid @property def visible(self) -> bool: """ Whether the Entity is visible in camera (checked in ANALYST object tree). """ return self._visible @visible.setter def visible(self, value: bool): self._visible = value self.workspace.update_attribute(self, "attributes") @property def workspace(self) -> Workspace: """ :obj:`~geoh5py.workspace.workspace.Workspace` to which the Entity belongs to. """ return self.entity_type.workspace