made timestamp usage more consistent, fixed grid metrics to use copies,

     made info and get_history have fewer required parameters

Author: kylemann16

src/silvimetric/cli/cli.py

@@ -150,12 +150,15 @@ def info_cmd(
 
     start_date = dates[0] if dates else date
     end_date = dates[1] if dates else date
+    if start_date is None and end_date is None:
+        timestamp=None
+    else:
+        timestamp = tuple(start_date.timestamp(), end_date.timestamp())
 
     i = info.info(
         app.tdb_dir,
         bounds=bounds,
-        start_time=start_date,
-        end_time=end_date,
+        timestamp=timestamp,
         name=name,
         concise=True,
     )
@@ -170,7 +173,6 @@ def info_cmd(
     ]
 
     i['metadata'].pop('metrics')
-    # print(metrics.keys())
     if any([history, metadata, attributes, metrics]):
         filtered = {}
         if history:

src/silvimetric/commands/info.py

@@ -6,48 +6,9 @@
 from typing import Optional
 
 
-def check_values(
-    start_time: datetime,
-    end_time: datetime,
-    bounds: Bounds,
-    name: Union[UUID, str],
-):
-    """
-    Validate arguments for info command.
-
-    :param start_time: Starting datetime object.
-    :param end_time: Ending datetime object.
-    :param bounds: Bounds to query by.
-    :param name: Name to query by.
-    :raises TypeError: Incorrect type of start_time argument.
-    :raises TypeError: Incorrect type of end_time argument.
-    :raises TypeError: Incorrect type of bounds argument.
-    :raises TypeError: Incorrect type of name argument.
-    :raises TypeError: Incorrect type of name argument.
-    :meta public:
-    """
-    if start_time is not None and not isinstance(start_time, datetime):
-        raise TypeError('Incorrect type of "start_time" argument.')
-    if end_time is not None and not isinstance(end_time, datetime):
-        raise TypeError('Incorrect type of "end_time" argument.')
-    if bounds is not None and not isinstance(bounds, Bounds):
-        raise TypeError('Incorrect type of "bounds" argument.')
-    if name is not None:
-        if isinstance(name, UUID):
-            pass
-        elif isinstance(name, str):
-            try:
-                UUID(name)
-            except Exception as e:
-                raise TypeError('Incorrect type of "name" argument.') from e
-        else:
-            raise TypeError('Incorrect type of "name" argument.')
-
-
 def info(
     tdb_dir: str,
-    start_time: Optional[datetime] = None,
-    end_time: Optional[datetime] = None,
+    timestamp: Optional[tuple[int, int]] = None,
     bounds: Optional[Bounds] = None,
     name: Optional[Union[str, UUID]] = None,
     concise: Optional[bool] = False,
@@ -62,13 +23,7 @@ def info(
     :param name: Name query, defaults to None
     :return: Returns json object containing information on database.
     """
-    check_values(start_time, end_time, bounds, name)
-
     with Storage.from_db(tdb_dir) as tdb:
-        if start_time is None:
-            start_time = datetime.fromtimestamp(0)
-        if end_time is None:
-            end_time = datetime.now()
         if bounds is None:
             bounds = tdb.config.root
         if name is not None:
@@ -85,10 +40,7 @@ def info(
         }
 
         try:
-            history = tdb.get_history(
-                start_time, end_time, bounds, name, concise
-            )
-
+            history = tdb.get_history(timestamp, bounds, name, concise)
             if bool(history) and isinstance(history, list):
                 history = [h for h in history]
             elif bool(history):

src/silvimetric/commands/scan.py

@@ -47,51 +47,53 @@ def scan(
         )
         logger.debug(json.dumps(thresholds, indent=2))
 
-        with ProgressBar():
-            extents = Extents.from_sub(tdb_dir, data.bounds)
-            logger.info('Gathering initial chunks...')
-            count = dask.delayed(data.estimate_count)(extents.bounds).persist()
-            cell_counts = extent_handle(
-                extents, data, resolution, point_count, depth, log
-            )
+        extents = Extents.from_sub(tdb_dir, data.bounds)
+        logger.info('Gathering initial chunks...')
+        count = dask.delayed(data.estimate_count)(extents.bounds).persist()
+        cell_counts = extent_handle(
+            extents, data, resolution, point_count, depth, log
+        )
 
-            np_cell_counts = np.array(cell_counts)
-            num_cells = np.sum(np_cell_counts).item()
+        np_cell_counts = np.array(cell_counts)
+        num_cells = np.sum(np_cell_counts).item()
+        if np_cell_counts.size > 1:
             q1, q3 = np.percentile(np_cell_counts, [25,75])
             iqr = q3 - q1
             low_bounds = q1 - (1.5 * iqr)
             up_bounds = q3 + (1.5 * iqr)
 
             adjusted = np_cell_counts[np_cell_counts > low_bounds]
             adjusted = adjusted[adjusted < up_bounds]
-
-            std = np.std(adjusted)
-            mean = np.mean(adjusted)
-            median = np.median(adjusted)
-            rec = median
-
-            pc_info = dict(
-                pc_info=dict(
-                    storage_bounds=tdb.config.root.to_json(),
-                    data_bounds=data.bounds.to_json(),
-                    count=dask.compute(count),
-                )
+        else:
+            adjusted = np_cell_counts
+
+        std = np.std(adjusted)
+        mean = np.mean(adjusted)
+        median = np.median(adjusted)
+        rec = median
+
+        pc_info = dict(
+            pc_info=dict(
+                storage_bounds=tdb.config.root.to_json(),
+                data_bounds=data.bounds.to_json(),
+                count=dask.compute(count),
             )
-            tiling_info = dict(
-                tile_info=dict(
-                    num_cells=num_cells,
-                    num_tiles=len(cell_counts),
-                    mean=mean,
-                    std_dev=std,
-                    median=median,
-                    recommended=rec,
-                )
+        )
+        tiling_info = dict(
+            tile_info=dict(
+                num_cells=num_cells,
+                num_tiles=len(cell_counts),
+                mean=mean,
+                std_dev=std,
+                median=median,
+                recommended=rec,
             )
+        )
 
-            final_info = pc_info | tiling_info
-            logger.info(json.dumps(final_info, indent=2))
+        final_info = pc_info | tiling_info
+        logger.info(json.dumps(final_info, indent=2))
 
-            return final_info
+        return final_info
 
 
 def extent_handle(

src/silvimetric/commands/shatter.py

@@ -30,8 +30,8 @@ def final(
     config.finished = finished
     storage.save_shatter_meta(config)
 
-    persist(delayed(storage.consolidate)(config.timestamp))
-    compute(delayed(storage.vacuum)())
+    storage.consolidate(config.timestamp)
+    storage.vacuum()
 
 
 def get_data(extents: Extents, filename: str, storage: Storage) -> pd.DataFrame:
@@ -44,16 +44,16 @@ def get_data(extents: Extents, filename: str, storage: Storage) -> pd.DataFrame:
     :return: Point data array from PDAL.
     """
 
-    attrs = [a.name for a in storage.get_attributes()]
+    attrs = [*[a.name for a in storage.get_attributes()], 'xi', 'yi']
     data = Data(filename, storage.config, bounds=extents.bounds)
     p = data.pipeline
-    data.execute()
+    data.execute(allowed_dims = [*attrs, 'X', 'Y'])
 
     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 = points.loc[points.X < extents.bounds.maxx][attrs]
 
     points.loc[:, 'xi'] = np.floor(points.xi)
     # ceil for y because origin is at top left
@@ -125,21 +125,20 @@ def write(
     return p
 
 
+@delayed
 def do_one(leaf: Extents, config: ShatterConfig, storage: Storage) -> db.Bag:
     """Create dask bags and the order of operations."""
 
-    timestamp = config.timestamp
-
     # remove any extents that have already been done, only skip if full overlap
     if config.mbr:
         if not all(leaf.disjoint_by_mbr(m) for m in config.mbr):
             return 0
 
-    points = delayed(get_data)(leaf, config.filename, storage)
-    listed_data = delayed(agg_list)(points, config.time_slot)
-    metric_data = delayed(run_graph)(points, storage.get_metrics())
-    joined_data = delayed(join)(listed_data, metric_data)
-    point_count = delayed(write)(joined_data, storage, timestamp)
+    points = get_data(leaf, config.filename, storage)
+    listed_data = agg_list(points, config.time_slot)
+    metric_data = run_graph(points, storage.get_metrics())
+    joined_data = join(listed_data, metric_data)
+    point_count = write(joined_data, storage, config.timestamp)
 
     return point_count
 
@@ -157,6 +156,9 @@ def run(leaves: Leaves, config: ShatterConfig, storage: Storage) -> int:
     :return: Number of points processed.
     """
     ## If dask is distributed, use the futures feature
+    leaves = [delayed(leaf) for leaf in leaves]
+    storage = delayed(storage)
+
     processes = [do_one(leaf, config, storage) for leaf in leaves]
     dc = get_client()
     if dc is not None:
@@ -221,19 +223,19 @@ def shatter(config: ShatterConfig) -> int:
     storage.save_shatter_meta(config)
 
     config.log.debug('Grabbing leaf nodes.')
-    es = extents.chunk(data, pc_threshold=(15 * 10**6))
+    es = extents.chunk(data, pc_threshold=(11 * 10**6))
 
     for e in es:
         if config.tile_size is not None:
             leaves = e.get_leaf_children(config.tile_size)
         else:
-            leaves = itertools.chain(e.chunk(data))
+            leaves = e.chunk(data, pc_threshold=(600*10**3))
 
         # Begin main operations
         config.log.debug('Shattering.')
         try:
             run(leaves, config, storage)
-            persist(delayed(storage.consolidate)(config.timestamp))
+            storage.consolidate(config.timestamp)
         except Exception as e:
             final(config, storage)
             raise e