afscgap.flat_index_util

  1"""
  2Utilities to describe and execute filters over precomputed indicies.
  3
  4Utilities to describe and execute filters over precomputed indicies which, provided in Avro, may
  5help avoid requesting unnecessary catches.
  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 itertools
 15import typing
 16
 17import afscgap.convert
 18import afscgap.param
 19
 20MATCH_TARGET = typing.Union[float, int, str, None]
 21STRS = typing.Iterable[str]
 22
 23
 24class IndexFilter:
 25    """Interface for a filter against a precomupted index."""
 26
 27    def __init__(self):
 28        """Create a new index filter."""
 29        raise NotImplementedError('Use implementor.')
 30
 31    def get_index_names(self) -> STRS:
 32        """Get the name of the precomputed index to use to filter results.
 33
 34        Returns:
 35            The name of the precomputed index which can be used to execute this filter.
 36        """
 37        raise NotImplementedError('Use implementor.')
 38
 39    def get_matches(self, target: MATCH_TARGET) -> bool:
 40        """Determine a value matches this filter.
 41
 42        Args:
 43            target: The value to test if matches this filter.
 44
 45        Returns:
 46            True if this matches this filter's critera for being included in results for False
 47            otherwise.
 48        """
 49        raise NotImplementedError('Use implementor.')
 50
 51
 52class StringEqIndexFilter(IndexFilter):
 53    """Precomputed index filter that checks for string equality."""
 54
 55    def __init__(self, index_name: str, param: afscgap.param.StrEqualsParam):
 56        """Create a new string equals filter.
 57
 58        Args:
 59            index_name: The name of the precomputed index filter to use for finding results.
 60            param: The string equals parameter to apply to the precomputed index.
 61        """
 62        self._index_name = index_name
 63        self._param = param
 64
 65    def get_index_names(self) -> STRS:
 66        return [self._index_name]
 67
 68    def get_matches(self, value) -> bool:
 69        return value is not None and value == self._param.get_value()
 70
 71
 72class StringRangeIndexFilter(IndexFilter):
 73    """Precomputed index filter that checks for string within alphanumeric range."""
 74
 75    def __init__(self, index_name: str, param: afscgap.param.StrRangeParam):
 76        """Create a new string range filter.
 77
 78        Args:
 79            index_name: The name of the precomputed index filter to use for finding results.
 80            param: The string range parameter to apply to the precomputed index.
 81        """
 82        self._index_name = index_name
 83        self._param = param
 84
 85    def get_index_names(self) -> STRS:
 86        return [self._index_name]
 87
 88    def get_matches(self, value) -> bool:
 89        if value is None:
 90            return False
 91
 92        if self._param.get_low() is not None:
 93            satisfies_low = value >= self._param.get_low()
 94        else:
 95            satisfies_low = True
 96
 97        if self._param.get_high() is not None:
 98            satisfies_high = value <= self._param.get_high()
 99        else:
100            satisfies_high = True
101
102        return satisfies_low and satisfies_high
103
104
105class IntEqIndexFilter(IndexFilter):
106    """Precomputed index filter that checks for integer equality."""
107
108    def __init__(self, index_name: str, param: afscgap.param.IntEqualsParam):
109        """Create a new integer equals filter.
110
111        Args:
112            index_name: The name of the precomputed index filter to use for finding results.
113            param: The integer equals parameter to apply to the precomputed index.
114        """
115        self._index_name = index_name
116        self._param = param
117
118    def get_index_names(self) -> STRS:
119        return [self._index_name]
120
121    def get_matches(self, value) -> bool:
122        return value is not None and value == self._param.get_value()
123
124
125class IntRangeIndexFilter(IndexFilter):
126    """Precomputed index filter that checks for an integer in a range."""
127
128    def __init__(self, index_name: str, param: afscgap.param.IntRangeParam):
129        """Create a new integer range filter.
130
131        Args:
132            index_name: The name of the precomputed index filter to use for finding results.
133            param: The integer range parameter to apply to the precomputed index.
134        """
135        self._index_name = index_name
136        self._param = param
137
138    def get_index_names(self) -> STRS:
139        return [self._index_name]
140
141    def get_matches(self, value) -> bool:
142        if value is None:
143            return False
144
145        if self._param.get_low() is not None:
146            satisfies_low = value >= self._param.get_low()
147        else:
148            satisfies_low = True
149
150        if self._param.get_high() is not None:
151            satisfies_high = value <= self._param.get_high()
152        else:
153            satisfies_high = True
154
155        return satisfies_low and satisfies_high
156
157
158class FloatEqIndexFilter(IndexFilter):
159    """Precomputed index filter that checks for float approximate equality."""
160
161    def __init__(self, index_name: str, param: afscgap.param.FloatEqualsParam):
162        """Create a new float approximate equals filter.
163
164        Args:
165            index_name: The name of the precomputed index filter to use for finding results.
166            param: The float equals parameter to apply to the precomputed index.
167        """
168        self._index_name = index_name
169        self._param = param
170        self._param_str = self._prep_string(self._param.get_value())
171
172    def get_index_names(self) -> STRS:
173        return [self._index_name]
174
175    def get_matches(self, target: MATCH_TARGET) -> bool:
176        value = self._prep_string(target)
177
178        if value is None:
179            return False
180        else:
181            return value == self._param_str
182
183    def _prep_string(self, target) -> typing.Optional[str]:
184        if target is None:
185            return None
186        else:
187            return '%.2f' % target  # type: ignore
188
189
190class FloatRangeIndexFilter(IndexFilter):
191    """Precomputed index filter that checks for an floating point value in a range.
192
193    Precomputed index filter that checks for an floating point value in a range, using an
194    approximation. This will require local filtering to apply precision.
195    """
196
197    def __init__(self, index_name: str, param: afscgap.param.FloatRangeParam):
198        """Create a new float approximate range filter.
199
200        Args:
201            index_name: The name of the precomputed index filter to use for finding results.
202            param: The float range parameter to apply to the precomputed index.
203        """
204        self._index_name = index_name
205        self._param = param
206        self._low_str = self._prep_string(self._param.get_low())
207        self._high_str = self._prep_string(self._param.get_high())
208
209    def get_index_names(self) -> STRS:
210        return [self._index_name]
211
212    def get_matches(self, target: MATCH_TARGET) -> bool:
213        value = self._prep_string(target)
214
215        if value is None:
216            return False
217
218        if self._low_str is not None:
219            satisfies_low = value >= self._low_str
220        else:
221            satisfies_low = True
222
223        if self._high_str is not None:
224            satisfies_high = value <= self._high_str
225        else:
226            satisfies_high = True
227
228        return satisfies_low and satisfies_high
229
230    def _prep_string(self, target) -> typing.Optional[str]:
231        """Get a string which matches approximation / rounding used in the precomputed index.
232
233        Args:
234            target: The value to be converted to the index approximation / rounding.
235
236        Returns:
237            String describing the approximation / rounding of the input value which would be found
238            in the precomputed index.
239        """
240        if target is None:
241            return None
242        else:
243            return '%.2f' % target  # type: ignore
244
245
246class DatetimeEqIndexFilter(IndexFilter):
247    """Precomputed index filter that checks for approximate datetime equality."""
248
249    def __init__(self, index_name: str, param: afscgap.param.FloatEqualsParam):
250        """Create a new datetime approximate equals filter.
251
252        Args:
253            index_name: The name of the precomputed index filter to use for finding results.
254            param: The float equals parameter to apply to the precomputed index.
255        """
256        self._index_name = index_name
257        self._param = param
258        self._param_str = self._prep_string(self._param.get_value())
259
260    def get_index_names(self) -> STRS:
261        return [self._index_name]
262
263    def get_matches(self, target: MATCH_TARGET) -> bool:
264        value = self._prep_string(target)
265
266        if value is None:
267            return False
268        else:
269            return value == self._param_str
270
271    def _prep_string(self, target) -> typing.Optional[str]:
272        """Get a string which matches approximation / rounding used in the precomputed index.
273
274        Args:
275            target: The value to be converted to the index approximation / rounding.
276
277        Returns:
278            String describing the approximation / rounding of the input value which would be found
279            in the precomputed index.
280        """
281        if target is None:
282            return None
283        else:
284            return target.split('T')[0]  # type: ignore
285
286
287class DatetimeRangeIndexFilter(IndexFilter):
288    """Precomputed index filter that checks for a datetime value in a range.
289
290    Precomputed index filter that checks for an datetime value in a range, using an approximation.
291    This will require local filtering to apply precision.
292    """
293
294    def __init__(self, index_name: str, param: afscgap.param.FloatRangeParam):
295        """Create a new datetime approximate range filter.
296
297        Args:
298            index_name: The name of the precomputed index filter to use for finding results.
299            param: The datetime range parameter to apply to the precomputed index.
300        """
301        self._index_name = index_name
302        self._param = param
303        self._low_str = self._prep_string(self._param.get_low())
304        self._high_str = self._prep_string(self._param.get_high())
305
306    def get_index_names(self) -> STRS:
307        return [self._index_name]
308
309    def get_matches(self, target: MATCH_TARGET) -> bool:
310        value = self._prep_string(target)
311
312        if value is None:
313            return False
314
315        if self._low_str is not None:
316            satisfies_low = value >= self._low_str
317        else:
318            satisfies_low = True
319
320        if self._high_str is not None:
321            satisfies_high = value <= self._high_str
322        else:
323            satisfies_high = True
324
325        return satisfies_low and satisfies_high
326
327    def _prep_string(self, target) -> typing.Optional[str]:
328        """Get a string which matches approximation / rounding used in the precomputed index.
329
330        Args:
331            target: The value to be converted to the index approximation / rounding.
332
333        Returns:
334            String describing the approximation / rounding of the input value which would be found
335            in the precomputed index.
336        """
337        if target is None:
338            return None
339        else:
340            return target.split('T')[0]  # type: ignore
341
342
343class UnitConversionIndexFilter(IndexFilter):
344    """Index filter decorator which performs a unit conversion prior to applying an inner filter."""
345
346    def __init__(self, inner: IndexFilter, user_units: str, system_units: str):
347        """Create a new decorator which applies a unit conversion prior to calling an inner filter.
348
349        Args:
350            inner: The underlying filter to decorate.
351            user_units: Units exepected by the inner filter.
352            system_units: Original units within the underlying data.
353        """
354        self._inner = inner
355        self._user_units = user_units
356        self._system_units = system_units
357
358    def get_index_names(self) -> typing.Iterable[str]:
359        return self._inner.get_index_names()
360
361    def get_matches(self, value: MATCH_TARGET) -> bool:
362        if value is None:
363            converted = None
364        else:
365            original = float(value)  # type: ignore
366            converted = afscgap.convert.convert(original, self._system_units, self._user_units)
367
368        return self._inner.get_matches(converted)
369
370
371class LogicalOrIndexFilter(IndexFilter):
372    """A composite index filter which applies a logical or between multiple inner filters."""
373
374    def __init__(self, inners: typing.List[IndexFilter]):
375        """Create a new logical or index filter.
376
377        Args:
378            inners: The filters to apply, reporting True if any match or False if none match.
379        """
380        self._inners = inners
381
382        names = itertools.chain(*map(lambda x: x.get_index_names(), self._inners))
383        names_unique = set(names)
384
385        if len(names_unique) == 0:
386            raise RuntimeError('Logical or index filter requires one or more index.')
387
388        self._names = list(names_unique)
389
390    def get_index_names(self) -> STRS:
391        return self._names
392
393    def get_matches(self, value: MATCH_TARGET) -> bool:
394        matches = map(lambda x: x.get_matches(value), self._inners)
395        return functools.reduce(lambda a, b: a or b, matches)
396
397
398STRATEGIES = {
399    'str': {
400        'equals': StringEqIndexFilter,
401        'range': StringRangeIndexFilter
402    },
403    'int': {
404        'equals': IntEqIndexFilter,
405        'range': IntRangeIndexFilter
406    },
407    'float': {
408        'equals': FloatEqIndexFilter,
409        'range': FloatRangeIndexFilter
410    },
411    'datetime': {
412        'equals': DatetimeEqIndexFilter,
413        'range': DatetimeRangeIndexFilter
414    }
415}
416
417INDICIES = {
418    'year': ['year'],
419    'srvy': ['srvy'],
420    'survey': ['survey'],
421    'stratum': ['stratum'],
422    'station': ['station'],
423    'vessel_name': ['vessel_name'],
424    'vessel_id': ['vessel_id'],
425    'date_time': ['date_time'],
426    'latitude_dd': ['latitude_dd_start', 'latitude_dd_end'],
427    'longitude_dd': ['longitude_dd_start', 'longitude_dd_end'],
428    'species_code': ['species_code'],
429    'common_name': ['common_name'],
430    'scientific_name': ['scientific_name'],
431    'taxon_confidence': ['taxon_confidence'],
432    'cpue_kgha': ['cpue_kgkm2'],
433    'cpue_kgkm2': ['cpue_kgkm2'],
434    'cpue_kg1000km2': ['cpue_kgkm2'],
435    'cpue_noha': ['cpue_nokm2'],
436    'cpue_nokm2': ['cpue_nokm2'],
437    'cpue_no1000km2': ['cpue_nokm2'],
438    'weight_kg': ['weight_kg'],
439    'count': ['count'],
440    'bottom_temperature_c': ['bottom_temperature_c'],
441    'surface_temperature_c': ['surface_temperature_c'],
442    'depth_m': ['depth_m'],
443    'distance_fished_km': ['distance_fished_km'],
444    'net_width_m': ['net_width_m'],
445    'net_height_m': ['net_height_m'],
446    'area_swept_ha': ['area_swept_km2'],
447    'duration_hr': ['duration_hr']
448}
449
450FIELD_CONVERSIONS = {
451    'cpue_kgha': {'user': 'kg/ha', 'system': 'kg/km2'},
452    'cpue_kg1000km2': {'user': 'kg1000/km2', 'system': 'kg/km2'},
453    'cpue_noha': {'user': 'no/ha', 'system': 'no/km2'},
454    'cpue_no1000km2': {'user': 'no1000/km2', 'system': 'no/km2'},
455    'area_swept_ha': {'user': 'ha', 'system': 'km2'}
456}
457
458FIELD_DATA_TYPE_OVERRIDES = {'date_time': 'datetime'}
459
460# These fields, when indexed, ignore zero values. If not presence only, these need to be included.
461PRESENCE_ONLY_FIELDS = {'species_code', 'common_name', 'scientific_name'}
462
463
464def decorate_filter(field: str, original: IndexFilter) -> IndexFilter:
465    """Decorate a filter for unit conversion or other preprocessing if required.
466
467    Args:
468        field: The name of the underlying field for which decoration should be applied.
469        original: The undeocrated index filter.
470
471    Returns:
472        Decorated filter if decoration was required or original if not.
473    """
474    if field not in FIELD_CONVERSIONS:
475        return original
476
477    conversion = FIELD_CONVERSIONS[field]
478    user_units = conversion['user']
479    system_units = conversion['system']
480    return UnitConversionIndexFilter(original, user_units, system_units)
481
482
483def determine_if_ignorable(field: str, param: afscgap.param.Param, presence_only: bool) -> bool:
484    """Determine if a field parameter is ignored for pre-filtering.
485
486    Determine if a field parameter is ignored for pre-filtering, turning it into a noop because
487    pre-filtering isn't possible or precomputed indicies are not available.
488
489    Args:
490        field: The name of the field for which filters should be made.
491        param: The parameter to apply for the field.
492        presence_only: Flag indicating if the query is for presence so zero inference records can be
493            excluded.
494
495    Returns:
496        True if ignorable and false otherwise.
497    """
498    if param.get_is_ignorable():
499        return True
500
501    # If the field index is presence only and this isn't a presence only request, the index must be
502    # ignored (cannot be used to pre-filter results).
503    zero_inference_required = not presence_only
504    field_index_excludes_zeros = field in PRESENCE_ONLY_FIELDS
505    if zero_inference_required and field_index_excludes_zeros:
506        return True
507
508    filter_type = param.get_filter_type()
509    if filter_type == 'empty':
510        return True
511
512    return False
513
514
515def make_filters(field: str, param: afscgap.param.Param,
516    presence_only: bool) -> typing.Iterable[IndexFilter]:
517    """Make filters for a field describing a backend-agnostic parameter.
518
519    Args:
520        field: The name of the field for which filters should be made.
521        param: The parameter to apply for the field.
522        presence_only: Flag indicating if the query is for presence so zero inference records can be
523            excluded.
524
525    Returns:
526        Iterable over filters which implement the given parameter for precomputed indicies. This may
527        be approximated such that all matching results are included in results but some results may
528        included may not match, requiring re-evaluation locally.
529    """
530    if determine_if_ignorable(field, param, presence_only):
531        return []
532
533    filter_type = param.get_filter_type()
534
535    if field in FIELD_DATA_TYPE_OVERRIDES:
536        data_type = FIELD_DATA_TYPE_OVERRIDES[field]
537    else:
538        data_type = param.get_data_type()
539
540    data_type_strategies = STRATEGIES.get(data_type, None)
541    if data_type_strategies is None:
542        raise RuntimeError('Could not find filter strategy for type %s.' % data_type)
543
544    init_strategy = data_type_strategies.get(filter_type, None)
545    if init_strategy is None:
546        raise RuntimeError('Could not find filter strategy for type %s.' % filter_type)
547
548    indicies = INDICIES.get(field, [])
549    if len(indicies) == 0:
550        return []
551
552    undecorated_filters = map(lambda x: init_strategy(x, param), indicies)
553    decorated_filters = map(lambda x: decorate_filter(field, x), undecorated_filters)
554    decorated_filters_realized = list(decorated_filters)
555    return [LogicalOrIndexFilter(decorated_filters_realized)]