changing to hilbert ordering and adding sort=False for groupby usage for potential speedup

Author: kylemann16

src/silvimetric/commands/shatter.py

@@ -25,39 +25,21 @@ def get_data(extents: Extents, filename: str, storage: Storage):
     :param storage: :class:`silvimetric.resources.storage.Storage` database.
     :return: Point data array from PDAL.
     """
+
+    attrs = [a.name for a in storage.get_attributes()]
     data = Data(filename, storage.config, bounds=extents.bounds)
     p = data.pipeline
     data.execute()
-    return p.get_dataframe(0)
-
-
-def arrange(points: pd.DataFrame, leaf, attrs: list[str]):
-    """
-    Arrange data to fit key-value TileDB input format.
-
-    :param data: Tuple of indices and point data array (xis, yis, data).
-    :param leaf: :class:`silvimetric.resources.extents.Extent` being used.
-    :param attrs: List of attribute names.
-    :raises Exception: Missing attribute error.
-    :return: None if no work is done, or a tuple of indices and rearranged data.
-    """
-    if points is None:
-        return None
-    if points.size == 0:
-        return None
 
-    points = points.loc[points.Y < leaf.bounds.maxy]
-    points = points.loc[points.Y >= leaf.bounds.miny]
-    points = points.loc[points.X >= leaf.bounds.minx]
-    points = points.loc[points.X < leaf.bounds.maxx, [*attrs, 'xi', 'yi']]
-
-    if points.size == 0:
-        return None
+    points = p.get_dataframe(0)
+    points = points.loc[points.Y < extents.bounds.maxy]
+    points = points.loc[points.Y >= extents.bounds.miny]
+    points = points.loc[points.X >= extents.bounds.minx]
+    points = points.loc[points.X < extents.bounds.maxx, [*attrs, 'xi', 'yi']]
 
     points.loc[:, 'xi'] = da.floor(points.xi)
     # ceil for y because origin is at top left
     points.loc[:, 'yi'] = da.ceil(points.yi)
-
     return points
 
 
@@ -84,7 +66,7 @@ def agg_list(data_in, proc_num):
     col_dtypes = {a: o for a in data_in.columns if a not in ['xi', 'yi']}
 
     coerced = data_in.astype(col_dtypes | xyi_dtypes)
-    gb = coerced.groupby(['xi', 'yi'])
+    gb = coerced.groupby(['xi', 'yi'], sort=False)
     listed = gb.agg(lambda x: np.array(x, old_dtypes[x.name]))
     counts_df = gb[first_col_name].agg('count').rename('count')
     listed = listed.join(counts_df)
@@ -97,12 +79,6 @@ def join(list_data: pd.DataFrame, metric_data):
     """
     Join the list data and metric DataFrames together.
     """
-    if list_data is None or metric_data is None:
-        return None
-
-    if isinstance(metric_data, Delayed):
-        metric_data = metric_data.compute()
-
     return list_data.join(metric_data).reset_index()
 
 
@@ -127,38 +103,27 @@ def write(data_in, storage, timestamp):
 Leaves = Generator[Extents, None, None]
 
 
-def get_processes(
-    leaves: Leaves, config: ShatterConfig, storage: Storage
+def do_one(
+    leaf: Extents, config: ShatterConfig, storage: Storage
 ) -> db.Bag:
     """Create dask bags and the order of operations."""
 
-    ## Handle dask bag transitions through work states
-    attrs = [a.name for a in storage.config.attrs]
     timestamp = config.timestamp
 
     # remove any extents that have already been done, only skip if full overlap
-    leaf_bag: db.Bag = db.from_sequence(leaves)
     if config.mbr:
+        if not all(leaf.disjoint_by_mbr(m) for m in config.mbr):
+            return 0
 
-        def mbr_filter(one: Extents):
-            return all(one.disjoint_by_mbr(m) for m in config.mbr)
+    points = get_data(leaf, config.filename, storage)
+    if points.empty:
+        return 0
+    metric_data = run_graph(points, storage.get_metrics())
+    listed_data = agg_list(points, config.time_slot)
+    joined_data = join(listed_data, metric_data)
+    point_count = write(joined_data, storage, timestamp)
 
-        leaf_bag = leaf_bag.filter(mbr_filter)
-
-    def pc_filter(d: pd.DataFrame):
-        if d is None:
-            return False
-        return not d.empty
-
-    points: db.Bag = leaf_bag.map(get_data, config.filename, storage)
-    arranged: db.Bag = points.map(arrange, leaf_bag, attrs)
-    filtered: db.Bag = arranged.filter(pc_filter)
-    metrics: db.Bag = filtered.map(run_graph, storage.config.metrics)
-    lists: db.Bag = filtered.map(agg_list, config.time_slot)
-    joined: db.Bag = lists.map(join, metrics)
-    writes: db.Bag = joined.map(write, storage, timestamp)
-
-    return writes
+    return point_count
 
 
 def run(leaves: Leaves, config: ShatterConfig, storage: Storage) -> int:
@@ -189,7 +154,8 @@ def kill_gracefully(signum, frame):
 
     signal.signal(signal.SIGINT, kill_gracefully)
 
-    processes = get_processes(leaves, config, storage)
+    leaf_bag: db.Bag = db.from_sequence(leaves)
+    processes = leaf_bag.map(do_one, config, storage)
 
     ## If dask is distributed, use the futures feature
     dc = get_client()

src/silvimetric/resources/metric.py

@@ -204,7 +204,7 @@ def do(self, data: pd.DataFrame, *args) -> pd.DataFrame:
             data = data[attrs]
 
         idxer = data[idx]
-        gb = data.groupby(idx)
+        gb = data.groupby(idx, sort=False)
 
         # Arguments come in as separate dataframes returned from previous
         # metrics deemed dependencies. If there are dependencies for this
@@ -221,8 +221,11 @@ def merge(left, right):
             merged_args = args
 
         # lambda method for use in dataframe aggregator
-        def runner(d):
-            return self.sanitize_and_run(d, idxer, merged_args)
+        def runner(d, idx=idxer, m_args=merged_args):
+            return self.sanitize_and_run(d, idx, m_args)
+
+        # def runner(values, index, m=merged_args):
+        #     return self._method(values, index, merged_args)
 
         # create map of current column name to tuple of new column name and
         # metric method

src/silvimetric/resources/metrics/p_moments.py

@@ -1,7 +1,7 @@
 import numpy as np
 
 from ..metric import Metric
-
+import pdb
 
 def m_mean(data, *args):
     m = data.mean()

src/silvimetric/resources/storage.py

@@ -131,6 +131,7 @@ def create(config: StorageConfig, ctx: tiledb.Ctx = None):
         # https://docs.tiledb.com/main/how-to/performance/performance-tips/summary-of-factors#allows-duplicates
         schema = tiledb.ArraySchema(
             domain=domain,
+            cell_order='hilbert',
             attrs=[
                 count_att,
                 proc_att,

tests/test_metrics.py

@@ -18,7 +18,7 @@ def test_dag(
         g = Graph(metrics).init()
         assert g.initialized
         assert all(n.initialized for n in g.nodes.values())
-        assert list(g.nodes.keys()) == [
+        assert not set(g.nodes.keys()) ^ set([
             'l1',
             'l2',
             'l3',
@@ -27,7 +27,8 @@ def test_dag(
             'lskewness',
             'lkurtosis',
             'lmombase',
-        ]
+        ])
+
         g.run(metric_data)
         for node in g.nodes.values():
             assert isinstance(node.results, pd.DataFrame)
@@ -40,6 +41,8 @@ def test_metrics(
         metric_data: pd.DataFrame,
         metric_data_results: pd.DataFrame,
     ):
+        # TODO fix this, changed to hilbert ordering and changed
+        # to not using sort in groupbys
         metrics = list(s.values())
         graph = Graph(metrics)
         metrics = graph.run(metric_data)

tests/test_shatter.py

@@ -65,8 +65,9 @@ def test_command(
         shatter_config: ShatterConfig,
         storage: Storage,
         test_point_count: int,
-        threaded_dask,
+        # threaded_dask,
     ):
+        os.environ['NUMBA_GDB_BINARY'] = '/usr/local/bin/gdb'
         shatter(shatter_config)
         base = 11 if storage.config.alignment == 'AlignToCenter' else 10
         maxy = storage.config.root.maxy