-
Notifications
You must be signed in to change notification settings - Fork 0
/
filters.py
executable file
·220 lines (173 loc) · 8.86 KB
/
filters.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
"""Provide filters for querying close approaches and limit the generated results.
The `create_filters` function produces a collection of objects that is used by
the `query` method to generate a stream of `CloseApproach` objects that match
all of the desired criteria. The arguments to `create_filters` are provided by
the main module and originate from the user's command-line options.
This function can be thought to return a collection of instances of subclasses
of `AttributeFilter` - a 1-argument callable (on a `CloseApproach`) constructed
from a comparator (from the `operator` module), a reference value, and a class
method `get` that subclasses can override to fetch an attribute of interest from
the supplied `CloseApproach`.
The `limit` function simply limits the maximum number of values produced by an
iterator.
You'll edit this file in Tasks 3a and 3c.
"""
import operator
from itertools import islice
class UnsupportedCriterionError(NotImplementedError):
"""A filter criterion is unsupported."""
class AttributeFilter:
"""A general superclass for filters on comparable attributes.
An `AttributeFilter` represents the search criteria pattern comparing some
attribute of a close approach (or its attached NEO) to a reference value. It
essentially functions as a callable predicate for whether a `CloseApproach`
object satisfies the encoded criterion.
It is constructed with a comparator operator and a reference value, and
calling the filter (with __call__) executes `get(approach) OP value` (in
infix notation).
Concrete subclasses can override the `get` classmethod to provide custom
behavior to fetch a desired attribute from the given `CloseApproach`.
"""
def __init__(self, op, value):
"""Construct a new `AttributeFilter` from an binary predicate and a reference value.
The reference value will be supplied as the second (right-hand side)
argument to the operator function. For example, an `AttributeFilter`
with `op=operator.le` and `value=10` will, when called on an approach,
evaluate `some_attribute <= 10`.
:param op: A 2-argument predicate comparator (such as `operator.le`).
:param value: The reference value to compare against.
"""
self.op = op
self.value = value
def __call__(self, approach):
"""Invoke `self(approach)`."""
return self.op(self.get(approach), self.value)
@classmethod
def get(cls, approach):
"""Get an attribute of interest from a close approach.
Concrete subclasses must override this method to get an attribute of
interest from the supplied `CloseApproach`.
:param approach: A `CloseApproach` on which to evaluate this filter.
:return: The value of an attribute of interest, comparable to `self.value` via `self.op`.
"""
raise UnsupportedCriterionError
def __repr__(self):
return f"{self.__class__.__name__}(op=operator.{self.op.__name__}, value={self.value})"
class DateFilter(AttributeFilter):
"""A filter that matches close approaches that occurred on a specific date.
The date is provided at construction, and the filter matches
close approaches that occurred on that date, based on the
approach's time attribute.
:param op: A comparison operator, such as operator.eq or operator.lt.
:param value: The date to compare against.
"""
@classmethod
def get(cls, approach):
return approach.time.date()
class DistanceFilter(AttributeFilter):
"""A filter that matches close approaches based on distance from Earth.
The distance is provided at construction, and the filter matches
close approaches whose distance from Earth matches the criterion,
based on the approach's distance attribute.
:param op: A comparison operator, such as operator.eq or operator.lt.
:param value: The distance to compare against.
"""
@classmethod
def get(cls, approach):
return approach.distance
class VelocityFilter(AttributeFilter):
"""A filter that matches close approaches based on relative velocity.
The velocity is provided at construction, and the filter matches
close approaches whose relative velocity matches the criterion,
based on the approach's velocity attribute.
:param op: A comparison operator, such as operator.eq or operator.lt.
:param value: The velocity to compare against.
"""
@classmethod
def get(cls, approach):
return approach.velocity
class DiameterFilter(AttributeFilter):
"""A filter that matches close approaches based on NEO's diameter.
The diameter is provided at construction, and the filter matches
close approaches whose NEO's diameter matches the criterion.
:param op: A comparison operator, such as operator.eq or operator.lt.
:param value: The diameter to compare against.
"""
@classmethod
def get(cls, approach):
return approach.neo.diameter
class HazardousFilter(AttributeFilter):
"""A filter that matches close approaches based on whether the NEO is hazardous.
The hazardous flag is provided at construction, and the filter matches
close approaches whose NEO's hazardous attribute matches the criterion.
:param op: A comparison operator, such as operator.eq or operator.lt.
:param value: The hazardous flag to compare against.
"""
@classmethod
def get(cls, approach):
return approach.neo.hazardous
def create_filters(
date=None,
start_date=None,
end_date=None,
distance_min=None,
distance_max=None,
velocity_min=None,
velocity_max=None,
diameter_min=None,
diameter_max=None,
hazardous=None,
):
"""Create a collection of filters from user-specified criteria.
Each of these arguments is provided by the main module with a value from the
user's options at the command line. Each one corresponds to a different type
of filter. For example, the `--date` option corresponds to the `date`
argument, and represents a filter that selects close approaches that occurred
on exactly that given date. Similarly, the `--min-distance` option
corresponds to the `distance_min` argument, and represents a filter that
selects close approaches whose nominal approach distance is at least that
far away from Earth. Each option is `None` if not specified at the command
line (in particular, this means that the `--not-hazardous` flag results in
`hazardous=False`, not to be confused with `hazardous=None`).
The return value must be compatible with the `query` method of `NEODatabase`
because the main module directly passes this result to that method. For now,
this can be thought of as a collection of `AttributeFilter`s.
:param date: A `date` on which a matching `CloseApproach` occurs.
:param start_date: A `date` on or after which a matching `CloseApproach` occurs.
:param end_date: A `date` on or before which a matching `CloseApproach` occurs.
:param distance_min: A minimum nominal approach distance for a matching `CloseApproach`.
:param distance_max: A maximum nominal approach distance for a matching `CloseApproach`.
:param velocity_min: A minimum relative approach velocity for a matching `CloseApproach`.
:param velocity_max: A maximum relative approach velocity for a matching `CloseApproach`.
:param diameter_min: A minimum diameter of the NEO of a matching `CloseApproach`.
:param diameter_max: A maximum diameter of the NEO of a matching `CloseApproach`.
:param hazardous: Whether the NEO of a matching `CloseApproach` is potentially hazardous.
:return: A collection of filters for use with `query`.
"""
# Construct a collection of filters.
filters = [
DateFilter(operator.eq, date) if date else None,
DateFilter(operator.ge, start_date) if start_date else None,
DateFilter(operator.le, end_date) if end_date else None,
DistanceFilter(operator.ge, distance_min) if distance_min else None,
DistanceFilter(operator.le, distance_max) if distance_max else None,
VelocityFilter(operator.ge, velocity_min) if velocity_min else None,
VelocityFilter(operator.le, velocity_max) if velocity_max else None,
DiameterFilter(operator.ge, diameter_min) if diameter_min else None,
DiameterFilter(operator.le, diameter_max) if diameter_max else None,
HazardousFilter(operator.eq, hazardous)
if isinstance(hazardous, bool)
else None,
]
return filters
def limit(iterator, n=None):
"""Produce a limited stream of values from an iterator.
If `n` is 0 or None, don't limit the iterator at all.
:param iterator: An iterator of values.
:param n: The maximum number of values to produce.
:yield: The first (at most) `n` values from the iterator.
"""
# Produce at most `n` values from the given iterator.
if n:
return islice(iterator, n)
return iterator