afscgap.flat_local_filter

Utilities to describe and execute filters over downloaded results.

Utilities to describe and execute filters over downloaded results, typically after approximate filters on precomputed indicies.

This file is part of afscgap released under the BSD 3-Clause License. See LICENSE.md.

View Source

  1"""
  2Utilities to describe and execute filters over downloaded results.
  3
  4Utilities to describe and execute filters over downloaded results, typically after approximate
  5filters on precomputed indicies.
  6
  7(c) 2025 Regents of University of California / The Eric and Wendy Schmidt Center
  8for Data Science and the Environment at UC Berkeley.
  9
 10This file is part of afscgap released under the BSD 3-Clause License. See
 11LICENSE.md.
 12"""
 13import functools
 14import typing
 15
 16import afscgap.flat_model
 17import afscgap.model
 18import afscgap.param
 19
 20from afscgap.flat_model import PARAMS_DICT
 21
 22
 23class LocalFilter:
 24    """Interface for a locally applied filter."""
 25
 26    def __init__(self):
 27        """Create a new local filter."""
 28        raise NotImplementedError('Use implementor.')
 29
 30    def matches(self, target: afscgap.model.Record) -> bool:
 31        """Determine if a value matches this filter.
 32
 33        Args:
 34            target: The record to check for inclusion in results given this filter.
 35
 36        Returns:
 37            True if the given record matches this filter so should be included in the results set
 38            and false otherwise.
 39        """
 40        raise NotImplementedError('Use implementor.')
 41
 42
 43class EqualsLocalFilter(LocalFilter):
 44    """A locally applied filter to check if an attribute matches an expected value."""
 45
 46    def __init__(self, accessor, value):
 47        """Create a new equality filter which is applied locally.
 48
 49        Args:
 50            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
 51                value to be tested.
 52            value: The expected value such that matching this value means that the record is
 53                included in the results.
 54        """
 55        self._accessor = accessor
 56        self._value = value
 57
 58    def matches(self, target: afscgap.model.Record) -> bool:
 59        candidate = self._accessor(target)
 60        return self._value == candidate
 61
 62
 63class RangeLocalFilter(LocalFilter):
 64    """A locally applied filter to check if an attribute falls within a range."""
 65
 66    def __init__(self, accessor, low_value, high_value):
 67        """Create a new locally applied range filter.
 68
 69        Args:
 70            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
 71                value to be tested.
 72            low_value: The minimum value such that values lower are excluded from the results set or
 73                None if no minimum should be enforced.
 74            high_value: The maximum value such that values lower are excluded from the results set
 75                or None if no maximum should be enforced.
 76        """
 77        self._accessor = accessor
 78        self._low_value = low_value
 79        self._high_value = high_value
 80
 81    def matches(self, target: afscgap.model.Record) -> bool:
 82        candidate = self._accessor(target)
 83
 84        if candidate is None:
 85            return False
 86
 87        satisfies_low = (self._low_value is None) or (candidate >= self._low_value)
 88        satisfies_high = (self._high_value is None) or (candidate <= self._high_value)
 89        return satisfies_low and satisfies_high
 90
 91
 92class LogicalAndLocalFilter(LocalFilter):
 93    """Filter which applies a logical and across one or more filters."""
 94
 95    def __init__(self, inner_filters: typing.List[LocalFilter]):
 96        """Create a new logical and local filter.
 97
 98        Create a new logical and local filter such that
 99
100        Args:
101            inner_filters: The filter to place into a logical and relationship.
102        """
103        self._inner_filters = inner_filters
104
105    def matches(self, target: afscgap.model.Record) -> bool:
106        individual_values = map(lambda x: x.matches(target), self._inner_filters)
107        return functools.reduce(lambda a, b: a and b, individual_values, True)
108
109
110ACCESSORS = {
111    'year': lambda x: x.get_year(),
112    'srvy': lambda x: x.get_srvy(),
113    'survey': lambda x: x.get_survey(),
114    'survey_id': lambda x: x.get_survey_id(),
115    'cruise': lambda x: x.get_cruise(),
116    'haul': lambda x: x.get_haul(),
117    'stratum': lambda x: x.get_stratum(),
118    'station': lambda x: x.get_station(),
119    'vessel_name': lambda x: x.get_vessel_name(),
120    'vessel_id': lambda x: x.get_vessel_id(),
121    'date_time': lambda x: x.get_date_time(),
122    'latitude_dd': lambda x: x.get_latitude(units='dd'),
123    'longitude_dd': lambda x: x.get_longitude(units='dd'),
124    'species_code': lambda x: x.get_species_code(),
125    'common_name': lambda x: x.get_common_name(),
126    'scientific_name': lambda x: x.get_scientific_name(),
127    'taxon_confidence': lambda x: x.get_taxon_confidence(),
128    'cpue_kgha': lambda x: x.get_cpue_weight_maybe(units='kg/ha'),
129    'cpue_kgkm2': lambda x: x.get_cpue_weight_maybe(units='kg/km2'),
130    'cpue_kg1000km2': lambda x: x.get_cpue_weight_maybe(units='kg1000/km2'),
131    'cpue_noha': lambda x: x.get_cpue_count_maybe(units='no/ha'),
132    'cpue_nokm2': lambda x: x.get_cpue_count_maybe(units='no/km2'),
133    'cpue_no1000km2': lambda x: x.get_cpue_count_maybe(units='no1000/km2'),
134    'weight_kg': lambda x: x.get_weight_maybe(units='kg'),
135    'count': lambda x: x.get_count_maybe(),
136    'bottom_temperature_c': lambda x: x.get_bottom_temperature_maybe(units='c'),
137    'surface_temperature_c': lambda x: x.get_surface_temperature_maybe(units='c'),
138    'depth_m': lambda x: x.get_depth(units='m'),
139    'distance_fished_km': lambda x: x.get_depth(units='km'),
140    'net_width_m': lambda x: x.get_net_width(units='m'),
141    'net_height_m': lambda x: x.get_net_height(units='m'),
142    'area_swept_ha': lambda x: x.get_area_swept(units='ha'),
143    'duration_hr': lambda x: x.get_duration(units='hr')
144}
145
146FILTER_STRATEGIES = {
147    'empty': lambda accessor, param: None,
148    'equals': lambda accessor, param: EqualsLocalFilter(accessor, param.get_value()),
149    'range': lambda accessor, param: RangeLocalFilter(accessor, param.get_low(), param.get_high())
150}
151
152
153def build_filter(params: PARAMS_DICT) -> LocalFilter:
154    """Build a filter which describes a set of parameters.
155
156    Args:
157        params: The parameters dictionary for which to build a local filter.
158
159    Returns:
160        New filter which implements the given parameters into a local filter.
161    """
162    params_flat = params.items()
163    params_keyed = map(lambda x: afscgap.param.FieldParam(x[0], x[1]), params_flat)
164    params_required = filter(
165        lambda x: not x.get_param().get_is_ignorable(),
166        params_keyed
167    )
168    individual_filters_maybe = map(
169        lambda x: build_individual_filter(x.get_field(), x.get_param()),
170        params_required
171    )
172    individual_filters = filter(lambda x: x is not None, individual_filters_maybe)
173    individual_filters_realized = list(individual_filters)
174    return LogicalAndLocalFilter(individual_filters_realized)  # type: ignore
175
176
177def build_individual_filter(field: str, param: afscgap.param.Param) -> typing.Optional[LocalFilter]:
178    """Create a single filter which helps implement a param dict into a local index filter.
179
180    Create a single filter which helps implement a param dict into a local index filter by operating
181    on a single attribute.
182
183    Args:
184        field: The name of the field for which a filter is being bulit.
185        param: The parameter to implement into a local filter.
186
187    Returns:
188        A local filter handling the given field.
189    """
190    filter_type = param.get_filter_type()
191
192    if field not in ACCESSORS:
193        raise RuntimeError('Unsupported or unknown field: %s' % field)
194
195    if filter_type not in FILTER_STRATEGIES:
196        raise RuntimeError('Unsupported filter type: %s' % filter_type)
197
198    accessor = ACCESSORS[field]
199    strategy = FILTER_STRATEGIES[filter_type]
200    return strategy(accessor, param)

class LocalFilter: View Source

24class LocalFilter:
25    """Interface for a locally applied filter."""
26
27    def __init__(self):
28        """Create a new local filter."""
29        raise NotImplementedError('Use implementor.')
30
31    def matches(self, target: afscgap.model.Record) -> bool:
32        """Determine if a value matches this filter.
33
34        Args:
35            target: The record to check for inclusion in results given this filter.
36
37        Returns:
38            True if the given record matches this filter so should be included in the results set
39            and false otherwise.
40        """
41        raise NotImplementedError('Use implementor.')

Interface for a locally applied filter.

LocalFilter() View Source

27    def __init__(self):
28        """Create a new local filter."""
29        raise NotImplementedError('Use implementor.')

Create a new local filter.

def matches(self, target: afscgap.model.Record) -> bool: View Source

31    def matches(self, target: afscgap.model.Record) -> bool:
32        """Determine if a value matches this filter.
33
34        Args:
35            target: The record to check for inclusion in results given this filter.
36
37        Returns:
38            True if the given record matches this filter so should be included in the results set
39            and false otherwise.
40        """
41        raise NotImplementedError('Use implementor.')

Determine if a value matches this filter.

Arguments:

target: The record to check for inclusion in results given this filter.

Returns:

True if the given record matches this filter so should be included in the results set and false otherwise.

class EqualsLocalFilter(LocalFilter): View Source

44class EqualsLocalFilter(LocalFilter):
45    """A locally applied filter to check if an attribute matches an expected value."""
46
47    def __init__(self, accessor, value):
48        """Create a new equality filter which is applied locally.
49
50        Args:
51            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
52                value to be tested.
53            value: The expected value such that matching this value means that the record is
54                included in the results.
55        """
56        self._accessor = accessor
57        self._value = value
58
59    def matches(self, target: afscgap.model.Record) -> bool:
60        candidate = self._accessor(target)
61        return self._value == candidate

A locally applied filter to check if an attribute matches an expected value.

EqualsLocalFilter(accessor, value) View Source

47    def __init__(self, accessor, value):
48        """Create a new equality filter which is applied locally.
49
50        Args:
51            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
52                value to be tested.
53            value: The expected value such that matching this value means that the record is
54                included in the results.
55        """
56        self._accessor = accessor
57        self._value = value

Create a new equality filter which is applied locally.

Arguments:

accessor: Strategy (function) which, when given a afscgap.model.Record, returns the value to be tested.
value: The expected value such that matching this value means that the record is included in the results.

def matches(self, target: afscgap.model.Record) -> bool: View Source

59    def matches(self, target: afscgap.model.Record) -> bool:
60        candidate = self._accessor(target)
61        return self._value == candidate

Determine if a value matches this filter.

Arguments:

target: The record to check for inclusion in results given this filter.

Returns:

True if the given record matches this filter so should be included in the results set and false otherwise.

class RangeLocalFilter(LocalFilter): View Source

64class RangeLocalFilter(LocalFilter):
65    """A locally applied filter to check if an attribute falls within a range."""
66
67    def __init__(self, accessor, low_value, high_value):
68        """Create a new locally applied range filter.
69
70        Args:
71            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
72                value to be tested.
73            low_value: The minimum value such that values lower are excluded from the results set or
74                None if no minimum should be enforced.
75            high_value: The maximum value such that values lower are excluded from the results set
76                or None if no maximum should be enforced.
77        """
78        self._accessor = accessor
79        self._low_value = low_value
80        self._high_value = high_value
81
82    def matches(self, target: afscgap.model.Record) -> bool:
83        candidate = self._accessor(target)
84
85        if candidate is None:
86            return False
87
88        satisfies_low = (self._low_value is None) or (candidate >= self._low_value)
89        satisfies_high = (self._high_value is None) or (candidate <= self._high_value)
90        return satisfies_low and satisfies_high

A locally applied filter to check if an attribute falls within a range.

RangeLocalFilter(accessor, low_value, high_value) View Source

67    def __init__(self, accessor, low_value, high_value):
68        """Create a new locally applied range filter.
69
70        Args:
71            accessor: Strategy (function) which, when given a afscgap.model.Record, returns the
72                value to be tested.
73            low_value: The minimum value such that values lower are excluded from the results set or
74                None if no minimum should be enforced.
75            high_value: The maximum value such that values lower are excluded from the results set
76                or None if no maximum should be enforced.
77        """
78        self._accessor = accessor
79        self._low_value = low_value
80        self._high_value = high_value

Create a new locally applied range filter.

Arguments:

accessor: Strategy (function) which, when given a afscgap.model.Record, returns the value to be tested.
low_value: The minimum value such that values lower are excluded from the results set or None if no minimum should be enforced.
high_value: The maximum value such that values lower are excluded from the results set or None if no maximum should be enforced.

def matches(self, target: afscgap.model.Record) -> bool: View Source

82    def matches(self, target: afscgap.model.Record) -> bool:
83        candidate = self._accessor(target)
84
85        if candidate is None:
86            return False
87
88        satisfies_low = (self._low_value is None) or (candidate >= self._low_value)
89        satisfies_high = (self._high_value is None) or (candidate <= self._high_value)
90        return satisfies_low and satisfies_high

Determine if a value matches this filter.

Arguments:

target: The record to check for inclusion in results given this filter.

Returns:

True if the given record matches this filter so should be included in the results set and false otherwise.

class LogicalAndLocalFilter(LocalFilter): View Source

 93class LogicalAndLocalFilter(LocalFilter):
 94    """Filter which applies a logical and across one or more filters."""
 95
 96    def __init__(self, inner_filters: typing.List[LocalFilter]):
 97        """Create a new logical and local filter.
 98
 99        Create a new logical and local filter such that
100
101        Args:
102            inner_filters: The filter to place into a logical and relationship.
103        """
104        self._inner_filters = inner_filters
105
106    def matches(self, target: afscgap.model.Record) -> bool:
107        individual_values = map(lambda x: x.matches(target), self._inner_filters)
108        return functools.reduce(lambda a, b: a and b, individual_values, True)

Filter which applies a logical and across one or more filters.

LogicalAndLocalFilter(inner_filters: List[LocalFilter]) View Source

 96    def __init__(self, inner_filters: typing.List[LocalFilter]):
 97        """Create a new logical and local filter.
 98
 99        Create a new logical and local filter such that
100
101        Args:
102            inner_filters: The filter to place into a logical and relationship.
103        """
104        self._inner_filters = inner_filters

Create a new logical and local filter.

Create a new logical and local filter such that

Arguments:

inner_filters: The filter to place into a logical and relationship.

def matches(self, target: afscgap.model.Record) -> bool: View Source

106    def matches(self, target: afscgap.model.Record) -> bool:
107        individual_values = map(lambda x: x.matches(target), self._inner_filters)
108        return functools.reduce(lambda a, b: a and b, individual_values, True)

Determine if a value matches this filter.

Arguments:

target: The record to check for inclusion in results given this filter.

Returns:

True if the given record matches this filter so should be included in the results set and false otherwise.

ACCESSORS = {'year': <function <lambda>>, 'srvy': <function <lambda>>, 'survey': <function <lambda>>, 'survey_id': <function <lambda>>, 'cruise': <function <lambda>>, 'haul': <function <lambda>>, 'stratum': <function <lambda>>, 'station': <function <lambda>>, 'vessel_name': <function <lambda>>, 'vessel_id': <function <lambda>>, 'date_time': <function <lambda>>, 'latitude_dd': <function <lambda>>, 'longitude_dd': <function <lambda>>, 'species_code': <function <lambda>>, 'common_name': <function <lambda>>, 'scientific_name': <function <lambda>>, 'taxon_confidence': <function <lambda>>, 'cpue_kgha': <function <lambda>>, 'cpue_kgkm2': <function <lambda>>, 'cpue_kg1000km2': <function <lambda>>, 'cpue_noha': <function <lambda>>, 'cpue_nokm2': <function <lambda>>, 'cpue_no1000km2': <function <lambda>>, 'weight_kg': <function <lambda>>, 'count': <function <lambda>>, 'bottom_temperature_c': <function <lambda>>, 'surface_temperature_c': <function <lambda>>, 'depth_m': <function <lambda>>, 'distance_fished_km': <function <lambda>>, 'net_width_m': <function <lambda>>, 'net_height_m': <function <lambda>>, 'area_swept_ha': <function <lambda>>, 'duration_hr': <function <lambda>>}

FILTER_STRATEGIES = {'empty': <function <lambda>>, 'equals': <function <lambda>>, 'range': <function <lambda>>}

def build_filter( params: Dict[str, afscgap.param.Param]) -> LocalFilter: View Source

154def build_filter(params: PARAMS_DICT) -> LocalFilter:
155    """Build a filter which describes a set of parameters.
156
157    Args:
158        params: The parameters dictionary for which to build a local filter.
159
160    Returns:
161        New filter which implements the given parameters into a local filter.
162    """
163    params_flat = params.items()
164    params_keyed = map(lambda x: afscgap.param.FieldParam(x[0], x[1]), params_flat)
165    params_required = filter(
166        lambda x: not x.get_param().get_is_ignorable(),
167        params_keyed
168    )
169    individual_filters_maybe = map(
170        lambda x: build_individual_filter(x.get_field(), x.get_param()),
171        params_required
172    )
173    individual_filters = filter(lambda x: x is not None, individual_filters_maybe)
174    individual_filters_realized = list(individual_filters)
175    return LogicalAndLocalFilter(individual_filters_realized)  # type: ignore

Build a filter which describes a set of parameters.

Arguments:

params: The parameters dictionary for which to build a local filter.

Returns:

New filter which implements the given parameters into a local filter.

def build_individual_filter( field: str, param: afscgap.param.Param) -> Optional[LocalFilter]: View Source

178def build_individual_filter(field: str, param: afscgap.param.Param) -> typing.Optional[LocalFilter]:
179    """Create a single filter which helps implement a param dict into a local index filter.
180
181    Create a single filter which helps implement a param dict into a local index filter by operating
182    on a single attribute.
183
184    Args:
185        field: The name of the field for which a filter is being bulit.
186        param: The parameter to implement into a local filter.
187
188    Returns:
189        A local filter handling the given field.
190    """
191    filter_type = param.get_filter_type()
192
193    if field not in ACCESSORS:
194        raise RuntimeError('Unsupported or unknown field: %s' % field)
195
196    if filter_type not in FILTER_STRATEGIES:
197        raise RuntimeError('Unsupported filter type: %s' % filter_type)
198
199    accessor = ACCESSORS[field]
200    strategy = FILTER_STRATEGIES[filter_type]
201    return strategy(accessor, param)

Create a single filter which helps implement a param dict into a local index filter.

Create a single filter which helps implement a param dict into a local index filter by operating on a single attribute.

Arguments:

field: The name of the field for which a filter is being bulit.
param: The parameter to implement into a local filter.

Returns:

A local filter handling the given field.