Dfview

exetera/core/abstract_types.py

@@ -65,6 +65,11 @@ def indexed(self):
     def data(self):
         raise NotImplementedError()
 
+    @property
+    @abstractmethod
+    def filter(self):
+        raise NotImplementedError()
+
     @abstractmethod
     def __bool__(self):
         raise NotImplementedError()
@@ -491,3 +496,30 @@ def ordered_merge_right(self, right_on, left_on,
                             left_field_sources=tuple(), right_field_sinks=None,
                             right_to_left_map=None, right_unique=False, left_unique=False):
         raise NotImplementedError()
+
+
+class SubjectObserver(ABC):
+    def attach(self, observer):
+        """
+        Attach the observer (view) to the subject (field).
+        """
+        raise NotImplementedError()
+
+    def detach(self, observer):
+        """
+        Detach the observer (view) from the subject (field), this is to remove the association between observer with subject.
+        """
+        raise NotImplementedError()
+
+    def notify(self, msg=None):
+        """
+        Called by the Subject to notify the observer on something.
+        """
+        raise NotImplementedError()
+
+    def update(self, subject, msg=None):
+        """
+        Called inside the observer, to perform actions based on subject and message type.
+        """
+        raise NotImplementedError()
+

exetera/core/dataframe.py

@@ -58,10 +58,16 @@ def __init__(self,
         self.name = name
         self._columns = OrderedDict()
         self._dataset = dataset
-        self._h5group = h5group
+        self._h5group = h5group  # the HDF5 group to store all fields
 
         for subg in h5group.keys():
-            self._columns[subg] = dataset.session.get(h5group[subg])
+            if subg[0] != '_':  # stores metadata, for example filters
+                self._columns[subg] = dataset.session.get(h5group[subg])
+
+        if '_filters' not in h5group.keys():
+            self._filters_grp = self._h5group.create_group('_filters')
+        else:
+            self._filters_grp = h5group['_filters']
 
     @property
     def columns(self):
@@ -101,15 +107,53 @@ def add(self,
             nfield.data.write(field.data[:])
         self._columns[dname] = nfield
 
+    def _add_view(self, field: fld.Field, filter: np.ndarray = None):
+        """
+        Internal function called by apply_filter to add a field view into the dataframe.
+
+        :param field: The field to apply filter to.
+        :param filter: The filter to apply.
+        :return: The field view.
+
+        """
+        # add view
+        view = type(field)(field._session, field._field, self, write_enabled=True)
+        field.attach(view)
+        self._columns[view.name] = view
+
+        # add filter
+        if filter is not None:
+            nformat = 'int32'
+            if len(filter) > 0 and np.max(filter) >= 2**31 - 1:
+                nformat = 'int64'
+            filter_name = view.name
+            if filter_name not in self._filters_grp.keys():
+                fld.numeric_field_constructor(self._dataset.session, self._filters_grp, filter_name, nformat)
+                filter_field = fld.NumericField(self._dataset.session, self._filters_grp[filter_name], self,
+                                                write_enabled=True)
+                filter_field.data.write(filter)
+            else:
+                filter_field = fld.NumericField(self._dataset.session, self._filters_grp[filter_name], self,
+                                                write_enabled=True)
+                if nformat not in filter_field._fieldtype:
+                    filter_field = filter_field.astype(nformat)
+                filter_field.data.clear()
+                filter_field.data.write(filter)
+
+            view._filter_wrapper = fld.ReadOnlyFieldArray(filter_field, 'values')  # read-only
+
+        return self._columns[view.name]
+
     def drop(self,
              name: str):
         """
         Drop a field from this dataframe as well as the HDF5 Group
 
         :param name: name of field to be dropped
         """
-        del self._columns[name]
-        del self._h5group[name]
+        del self._columns[name]  # should always be
+        if name in self._h5group.keys():  # in case of reference only
+            del self._h5group[name]
 
     def create_group(self,
                      name: str):
@@ -317,8 +361,10 @@ def __delitem__(self, name):
         if not self.__contains__(name=name):
             raise ValueError("There is no field named '{}' in this dataframe".format(name))
         else:
-            del self._h5group[name]
-            del self._columns[name]
+            del self._columns[name]  # should always be
+            if name in self._h5group.keys():  # in case of reference only
+                del self._h5group[name]
+
 
     def delete_field(self, field):
         """
@@ -478,13 +524,12 @@ def apply_filter(self, filter_to_apply, ddf=None):
         :returns: a dataframe contains all the fields filterd, self if ddf is not set
         """
         filter_to_apply_ = val.validate_filter(filter_to_apply)
-
-        if ddf is not None:
+        if ddf is not None and ddf is not self:
             if not isinstance(ddf, DataFrame):
                 raise TypeError("The destination object must be an instance of DataFrame.")
+            filter_to_apply_ = filter_to_apply_.nonzero()[0]
             for name, field in self._columns.items():
-                newfld = field.create_like(ddf, name)
-                field.apply_filter(filter_to_apply_, target=newfld)
+                ddf._add_view(field, filter_to_apply_)
             return ddf
         else:
             for field in self._columns.values():
@@ -514,15 +559,16 @@ def apply_index(self, index_to_apply, ddf=None):
         :param ddf: optional- the destination data frame
         :returns: a dataframe contains all the fields re-indexed, self if ddf is not set
         """
-        if ddf is not None:
+        if ddf is not None and ddf is not self:
             if not isinstance(ddf, DataFrame):
                 raise TypeError("The destination object must be an instance of DataFrame.")
             for name, field in self._columns.items():
-                newfld = field.create_like(ddf, name)
-                field.apply_index(index_to_apply, target=newfld)
+                # newfld = field.create_like(ddf, name)
+                # field.apply_index(index_to_apply, target=newfld)
+                ddf._add_view(field, index_to_apply)
             return ddf
         else:
-            val.validate_all_field_length_in_df(self) 
+            val.validate_all_field_length_in_df(self)
 
             for field in self._columns.values():
                 field.apply_index(index_to_apply, in_place=True)
@@ -981,6 +1027,11 @@ def describe(self, include=None, exclude=None, output='terminal'):
                 print('\n')
         return result
 
+    def view(self):
+        dfv = self.dataset.create_dataframe(self.name + '_view')
+        for f in self.columns.values():
+            dfv._add_view(f)
+        return dfv
 
 
 class HDF5DataFrameGroupBy(DataFrameGroupBy):
@@ -1656,4 +1707,4 @@ def _ordered_merge(left: DataFrame,
         if right[k].indexed:
             ops.ordered_map_valid_indexed_stream(right[k], right_map, dest_f, invalid)
         else:
-            ops.ordered_map_valid_stream(right[k], right_map, dest_f, invalid)
+            ops.ordered_map_valid_stream(right[k], right_map, dest_f, invalid)