"""Module containing occurrence data wranglers for filtering points."""
from logging import DEBUG
from lmpy.data_wrangling.occurrence.base import _OccurrenceDataWrangler
# .....................................................................................
[docs]class MinimumPointsFilter(_OccurrenceDataWrangler):
"""Filter points if there are less than the specified number."""
[docs] name = 'MinimumPointsWrangler'
# .......................
def __init__(self, minimum_count, **params):
"""Get an occurrence data wrangler for minimum points.
Args:
minimum_count (int): The minimum number of points in order to keep all.
**params (dict): Extra parameters to be sent to the base class.
"""
self.minimum_count = minimum_count
_OccurrenceDataWrangler.__init__(self, **params)
# .......................
[docs] def _get_all_pass_condition(self):
def all_pass_condition(point):
return True
return all_pass_condition
# .......................
[docs] def _get_all_fail_condition(self):
def all_fail_condition(point):
return False
return all_fail_condition
# .......................
[docs] def wrangle_points(self, points):
"""Wrangle points. Return all or none based on how many are present.
Args:
points (list of Point): A list of points to assess.
Returns:
list of Point: Either an empty list or all of the points passed in.
"""
try:
spname = points[0].species_name
except Exception:
spname = "unable to retrieve species name"
if len(points) >= self.minimum_count:
# Enough points, so all pass
self._pass_condition = self._get_all_pass_condition()
else:
# Not enough points to all fail
self._pass_condition = self._get_all_fail_condition()
self.logger.log(
f"{spname} {len(points)} fails minimum {self.minimum_count} points.",
refname=self.__class__.__name__, log_level=DEBUG)
# Call parent class wrangle_points function
return _OccurrenceDataWrangler.wrangle_points(self, points)