"""Filtering and masking operations for 3D point clouds."""
__author__ = "Yeremia Gunawan Adhisantoso"
__license__ = "Clear BSD"
__email__ = "adhisant@tnt.uni-hannover.de"
import typing as t
from pydantic import validate_call, ConfigDict
import numpy as np
import pandas as pd
from ... import consts as cm_consts
from ..matrices.rc_filters import filter_empty_rowcols
from ...utils.validate import require_positive_int
[docs]@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
def filter_points(
points: np.ndarray,
ret_mask: bool = False,
) -> np.ndarray | tuple[np.ndarray, np.ndarray]:
"""Filter out points with any NaN values.
Notes
-----
- If `ret_mask` is True, the function returns both the filtered points and the mask used for filtering.
- If `ret_mask` is False, only the filtered points are returned.
Parameters
----------
points : np.ndarray
The array of points to be filtered.
ret_mask : bool, optional
Whether to return the mask used for filtering, by default False.
Returns
-------
np.ndarray or Tuple[np.ndarray, np.ndarray]
The filtered points, and optionally the mask used for filtering.
"""
mask = ~np.isnan(points).any(axis=1)
new_points = points[mask, :]
if ret_mask:
return new_points, mask
else:
return points
[docs]@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
def filter_valid_points(
points: np.ndarray,
cm_df: pd.DataFrame,
ds_ratio: int = 1,
) -> np.ndarray:
"""Filter valid points based on the provided DataFrame and downsampling ratio.
This function is used when the coordinates of the points covers also the empty regions.
Notes
-----
- The function ensures that the `ds_ratio` is a positive integer.
- It extracts unique row and column IDs from the DataFrame and filters the points accordingly.
- If `ds_ratio` is greater than 1, it performs downsampling by averaging points within the same coarse-grained bin ID.
Parameters
----------
points : np.ndarray
The array of points to be filtered.
cm_df : pd.DataFrame
The DataFrame containing row and column IDs.
ds_ratio : int, optional
The downsampling ratio, by default 1. Must be a positive integer.
Returns
-------
np.ndarray
The filtered and optionally downsampled points.
"""
require_positive_int("ds_ratio", ds_ratio)
row_ids = cm_df[cm_consts.DataFrameSpecs.ROW_IDS].to_numpy()
col_ids = cm_df[cm_consts.DataFrameSpecs.COL_IDS].to_numpy()
out = filter_empty_rowcols((row_ids, col_ids), is_triu_sym=True, ret_unique_ids=True)
unique_ids = out[2]
max_id = unique_ids.max() if len(unique_ids) > 0 else 0
num_points = points.shape[0]
if max_id >= num_points:
raise ValueError(f"Max ID {max_id} is out of bounds for points array of shape {points.shape}")
new_points = points[unique_ids, :]
if ds_ratio > 1:
new_points = (
pd.DataFrame(data={
'lr_ids': unique_ids // ds_ratio,
'x': new_points[:, 0],
'y': new_points[:, 1],
'z': new_points[:, 2]
})
.groupby(['lr_ids'])
.mean()
.reset_index()
[['x', 'y', 'z']]
.to_numpy()
)
return new_points
[docs]@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
def mask_points(
points: np.ndarray,
cm_df: pd.DataFrame,
ds_ratio: int = 1,
) -> np.ndarray:
"""Masks points based on the provided DataFrame and downsampling ratio.
Notes
-----
This function processes the input points and masks them based on the unique
row and column IDs from the DataFrame. If the downsampling ratio is greater
than 1, it further processes the points to downsample them.
Parameters
----------
points : np.ndarray
The array of points to be masked.
cm_df : pd.DataFrame
The DataFrame containing row and column IDs.
ds_ratio : int, optional
The downsampling ratio, by default 1. Must be an integer greater than or equal to 1.
Returns
-------
np.ndarray
The masked points array.
"""
require_positive_int("ds_ratio", ds_ratio)
row_ids = cm_df[cm_consts.DataFrameSpecs.ROW_IDS].to_numpy()
col_ids = cm_df[cm_consts.DataFrameSpecs.COL_IDS].to_numpy()
out = filter_empty_rowcols((row_ids, col_ids), is_triu_sym=True, ret_unique_ids=True)
unique_ids = out[2]
max_id = unique_ids.max() if len(unique_ids) > 0 else 0
num_points, ndim = points.shape
if max_id >= num_points:
raise ValueError(f"Max ID {max_id} is out of bounds for points array of shape {points.shape}")
if ds_ratio == 1:
new_points = points.copy()
mask = np.ones(num_points, dtype=bool)
mask[unique_ids] = False
new_points[mask, :] = np.inf
elif ds_ratio > 1:
new_points = points[unique_ids, :]
new_num_points = int(np.ceil(num_points / ds_ratio))
lr_points_df = (
pd.DataFrame(data={
'lr_ids': unique_ids // ds_ratio,
'x': new_points[:, 0],
'y': new_points[:, 1],
'z': new_points[:, 2]
})
.groupby(['lr_ids'])
.mean()
.reset_index()
)
lr_ids = lr_points_df['lr_ids'].to_numpy()
lr_points = lr_points_df[['x', 'y', 'z']].to_numpy()
new_points = np.full((new_num_points, 3), np.inf)
new_points[lr_ids, :] = lr_points
return new_points