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,18 @@ 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):
+        raise NotImplementedError()
+
+    def detach(self, observer):
+        raise NotImplementedError()
+
+    def notify(self, msg=None):
+        raise NotImplementedError()
+
+    def update(self, subject, msg=None):
+        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,83 @@ 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
+        if isinstance(field, fld.NumericField):
+            view = fld.NumericField(field._session, field._field, self, write_enabled=True)
+        elif isinstance(field, fld.CategoricalField):
+            view = fld.CategoricalField(field._session, field._field, self, write_enabled=True)
+        elif isinstance(field, fld.TimestampField):
+            view = fld.TimestampField(field._session, field._field, self, write_enabled=True)
+        elif isinstance(field, fld.FixedStringField):
+            view = fld.FixedStringField(field._session, field._field, self, write_enabled=True)
+        elif isinstance(field, fld.IndexedStringField):
+            view = fld.IndexedStringField(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 filter[-1] < 2 ** 31 - 1 else '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 = filter_field.data
+
+        return self._columns[view.name]
+
+    # def change_filter(self, field: fld.Field, filter: np.ndarray):
+    #     """
+    #
+    #     :param field:
+    #     :param filter:
+    #     :return:
+    #     """
+    #     pass
+    #
+    # def remove_filter(self, field: Union[str, fld.Field]):
+    #     """
+    #     Remove filter from this dataframe specified by the field or field name.
+    #     """
+    #     if not isinstance(field, str) and not isinstance(field, fld.Field):
+    #         raise TypeError("The target field should be type field or string (name of the field in this dataframe).")
+    #
+    #     name = field if isinstance(field, str) else field.name
+    #     if name not in self._columns:
+    #         raise ValueError("The target field is not in this dataframe.")
+    #     else:
+    #         del self._filters_grp[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):
@@ -264,6 +338,14 @@ def contains_field(self, field):
                     return True
             return False
 
+
+    def _get_filter_grp(self, field: Union[str, fld.Field]=None):
+        """
+        Get a filter array specified by the field or field name.
+        """
+        filter_name = '_filter'
+        return self._filters_grp[filter_name]
+
     def __getitem__(self, name):
         """
         Get a field stored by the field name.
@@ -317,8 +399,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 +562,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 +597,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 +1065,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 +1745,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)