6534953321

Committed 16 Oct 2023 02:19PM UTC coverage: 35.674% (-22.7%) from 58.421%

Build # 6534953321

Build Type

push

github

Committed by

feihoo87

Commit Message

fix Coveralls

Run Details

5913 of 16575 relevant lines covered (35.67%)

3.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.63

/waveforms/math/interval.py

import ast
import bisect
import functools

try:
    from typing import Self
except:
    from typing import TypeVar
    Self = TypeVar('Self')

import numpy as np
from pyparsing import Forward, Literal, Optional, Or, Regex, Suppress

LCLOSED = Literal("[").setParseAction(lambda: [True])
LOPEN = Literal("(").setParseAction(lambda: [False])
RCLOSED = Literal("]").setParseAction(lambda: [True])
ROPEN = Literal(")").setParseAction(lambda: [False])
inf = Literal('inf').setParseAction(lambda: [np.inf])
number = Regex(r"[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?").setParseAction(
    lambda tokens: [ast.literal_eval(tokens[0])])

left = (LOPEN + Suppress('-') + inf).setParseAction(lambda: [False, -np.inf]) \
     | Or([LCLOSED, LOPEN]) + number

right = (Optional(Suppress('+')) + inf + ROPEN).setParseAction(lambda: [np.inf, False]) \
      | number + Or([RCLOSED, ROPEN])

interval = left + Suppress(",") + right
interval.setParseAction(lambda t: _Interval(t[1], t[2], t[0], t[3]))

expression = Forward()
expression << interval + Optional(Suppress("U") + expression)


@functools.total_ordering
class _Interval():

    __slots__ = ['start', 'end', 'start_closed', 'end_closed']

    def __init__(self,
                 start=-np.inf,
                 end=np.inf,
                 start_closed=False,
                 end_closed=False):
        assert start <= end, f"start {start} should not be greater than end {end}."
        if start == -np.inf:
            assert not start_closed, f"start {start} should not be closed."
        if end == np.inf:
            assert not end_closed, f"end {end} should not be closed."
        self.start = start
        self.end = end
        self.start_closed = start_closed
        self.end_closed = end_closed

    def __contains__(self, point) -> bool:
        if self.start < point < self.end:
            return True
        if point == self.start and self.start_closed:
            return True
        if point == self.end and self.end_closed:
            return True
        return False

    def is_subset_of(self, other: Self) -> bool:
        return ((other.start < self.start or other.start == self.start and
                 (other.start_closed or not self.start_closed))
                and (self.end < other.end or self.end == other.end and
                     (not self.end_closed or other.end_closed)))

    def intersects(self, other: Self):
        if self.start > other.end or self.end < other.start:
            return False
        if self.start == other.end and not (self.start_closed
                                            and other.end_closed):
            return False
        if self.end == other.start and not (self.end_closed
                                            and other.start_closed):
            return False
        return True

    def intersection(self, other: Self) -> Self:
        if not self.intersects(other):
            return None
        start, start_closed = max((self.start, not self.start_closed),
                                  (other.start, not other.start_closed))
        start_closed = not start_closed
        end, end_closed = min((self.end, self.end_closed),
                              (other.end, other.end_closed))

        return _Interval(start,
                         end,
                         start_closed=start_closed,
                         end_closed=end_closed)

    def union(self, other: Self) -> Self:
        if not self.intersects(other):
            return None
        start, start_closed = min((self.start, not self.start_closed),
                                  (other.start, not other.start_closed))
        start_closed = not start_closed
        end, end_closed = max((self.end, self.end_closed),
                              (other.end, other.end_closed))
        return _Interval(start,
                         end,
                         start_closed=start_closed,
                         end_closed=end_closed)

    def empty(self) -> bool:
        return self.start > self.end or (
            self.start == self.end
            and not (self.start_closed and self.end_closed))

    def __neg__(self) -> Self:
        return _Interval(-self.end, -self.start, self.end_closed,
                         self.start_closed)

    def __add__(self, other: int | float) -> Self:
        return _Interval(self.start + other, self.end + other,
                         self.start_closed, self.end_closed)

    def __sub__(self, other: int | float) -> Self:

        return _Interval(self.start - other, self.end - other,
                         self.start_closed, self.end_closed)

    def __mul__(self, other: int | float) -> Self:
        return _Interval(self.start * other, self.end * other,
                         self.start_closed, self.end_closed)

    def __truediv__(self, other: int | float) -> Self:
        return _Interval(self.start / other, self.end / other,
                         self.start_closed, self.end_closed)

    def __and__(self, other: Self) -> Self:
        return self.intersection(other)

    def __or__(self, other: Self) -> Self:
        return self.union(other)

    def __eq__(self, other):
        if isinstance(other, (int, float)):
            return not self.empty() and self.start == self.end == other
        if not isinstance(other, _Interval):
            raise TypeError(
                f"Cannot compare interval with {type(other)} {other}.")
        return (self.start == other.start and self.end == other.end
                and self.start_closed == other.start_closed
                and self.end_closed == other.end_closed)

    def __gt__(self, other):
        if isinstance(other, (int, float)):
            return self.start > other
        if not isinstance(other, _Interval):
            raise TypeError(
                f"Cannot compare interval with {type(other)} {other}.")
        return (self.start, not self.start_closed, self.end,
                not self.end_closed) > (other.start, not other.start_closed,
                                        other.end, not other.end_closed)

    def __str__(self) -> str:
        if self.start == self.end and self.start_closed and self.end_closed:
            return f"{{{self.start}}}"
        elif self.start >= self.end:
            return "{}"
        return f"{'[' if self.start_closed else '('}{self.start}, {self.end}{']' if self.end_closed else ')'}"

    def __repr__(self) -> str:
        return f"_Interval({self.start}, {self.end}, {self.start_closed}, {self.end_closed})"


class Interval():

    def __init__(self, *args):
        if len(args) == 1:
            if isinstance(args[0], list):
                intervals = args[0]
            elif isinstance(args[0], str):
                intervals = [
                    interval.parseString(s)[0] for s in args[0].split('U')
                ]
            else:
                raise ValueError(
                    'Interval can only be initialized with a list of intervals or a string'
                )
        elif len(args) == 2 and isinstance(args[0],
                                           (int, float)) and isinstance(
                                               args[1], (int, float)):
            intervals = [_Interval(*args)]
        elif len(args) == 4 and isinstance(args[0],
                                           (int, float)) and isinstance(
                                               args[1], (int, float)):
            intervals = [_Interval(*args)]
        else:
            intervals = []
        self.intervals: list[_Interval] = sorted(
            [interval for interval in intervals if not interval.empty()])

    def empty(self) -> bool:
        return len(self.intervals) == 0

    def full(self) -> bool:
        return len(self.intervals) == 1 and self.intervals[
            0].start == -np.inf and self.intervals[0].end == np.inf

    def __contains__(self, item: int | float) -> bool:
        if isinstance(item, (int, float)):
            i = bisect.bisect_left(self.intervals, item)
            if i == 0 or i > len(self.intervals):
                return False
            return item in self.intervals[i - 1]
        else:
            raise TypeError(f"Cannot check if {item} is in interval.")

    def is_subset_of(self, other: Self) -> bool:
        if self.empty():
            return True
        if other.empty():
            return False
        for interval in self.intervals:
            if not any(
                    interval.is_subset_of(other_interval)
                    for other_interval in other.intervals):
                return False
        return True

    def __eq__(self, other):
        if isinstance(other, Interval):
            return self.intervals == other.intervals
        elif isinstance(other, _Interval):
            return self.intervals == [other]
        else:
            raise TypeError(
                f"Cannot compare interval with {type(other)} {other}.")

    def __lt__(self, other: Self) -> bool:
        return self.is_subset_of(other) and not self == other

    def __le__(self, other: Self) -> bool:
        return self.is_subset_of(other)

    def __gt__(self, other: Self) -> bool:
        return other.is_subset_of(self) and not self == other

    def __ge__(self, other: Self) -> bool:
        return other.is_subset_of(self)

    def __neg__(self) -> Self:
        return Interval([-interval for interval in self.intervals])

    def __add__(self, other: int | float) -> Self:
        if isinstance(other, (int, float)):
            return Interval([interval + other for interval in self.intervals])
        else:
            raise TypeError(f"Cannot add interval with {type(other)} {other}.")

    def __radd__(self, other: int | float) -> Self:
        return self + other

    def __sub__(self, other: int | float) -> Self:
        if isinstance(other, (int, float)):
            return Interval([interval - other for interval in self.intervals])
        else:
            raise TypeError(f"Cannot add interval with {type(other)} {other}.")

    def __mul__(self, other: int | float) -> Self:
        if isinstance(other, (int, float)):
            return Interval([interval * other for interval in self.intervals])
        else:
            raise TypeError(f"Cannot add interval with {type(other)} {other}.")

    def __rmul__(self, other: int | float) -> Self:
        return self * other

    def __truediv__(self, other: int | float) -> Self:
        if isinstance(other, (int, float)):
            return Interval([interval / other for interval in self.intervals])
        else:
            raise TypeError(f"Cannot add interval with {type(other)} {other}.")

    def __invert__(self):
        compl_intervals = []
        prev_end = -np.inf
        prev_end_closed = False

        for interval in self.intervals:
            compl_intervals.append(
                _Interval(prev_end, interval.start, prev_end_closed,
                          not interval.start_closed))
            prev_end = interval.end
            prev_end_closed = not interval.end_closed

        compl_intervals.append(
            _Interval(prev_end, np.inf, prev_end_closed, False))
        return Interval(compl_intervals)

    def __and__(self, other):
        intersections = []
        for i in self.intervals:
            for j in other.intervals:
                intersection = i.intersection(j)
                if intersection is not None:
                    intersections.append(intersection)
        return Interval(intersections)

    def __or__(self, other):
        union = sorted(self.intervals + other.intervals)
        merged = [union[0]]
        for current in union:
            last = merged[-1]
            if current.start < last.end or (current.start == last.end and
                                            (last.end_closed
                                             or current.start_closed)):
                if current.end > last.end or (current.end == last.end
                                              and current.end_closed):
                    last.end = current.end
                    last.end_closed = current.end_closed
            else:
                merged.append(current)
        return Interval(merged)

    def __repr__(self):
        if len(self.intervals) == 0:
            return '{}'
        elif len(self.intervals) == 1:
            return str(self.intervals[0])
        else:
            return ' U '.join([str(x) for x in self.intervals])

1	import ast	9✔
2	import bisect	9✔
3	import functools	9✔
4
5	try:	9✔
6	from typing import Self	9✔
7	except:	3✔
8	from typing import TypeVar	3✔
9	Self = TypeVar('Self')	3✔
10
11	import numpy as np	9✔
12	from pyparsing import Forward, Literal, Optional, Or, Regex, Suppress	9✔
13
14	LCLOSED = Literal("[").setParseAction(lambda: [True])	9✔
15	LOPEN = Literal("(").setParseAction(lambda: [False])	9✔
16	RCLOSED = Literal("]").setParseAction(lambda: [True])	9✔
17	ROPEN = Literal(")").setParseAction(lambda: [False])	9✔
18	inf = Literal('inf').setParseAction(lambda: [np.inf])	9✔
19	number = Regex(r"[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?").setParseAction(	9✔
20	lambda tokens: [ast.literal_eval(tokens[0])])
21
22	left = (LOPEN + Suppress('-') + inf).setParseAction(lambda: [False, -np.inf]) \	9✔
23	\| Or([LCLOSED, LOPEN]) + number
24
25	right = (Optional(Suppress('+')) + inf + ROPEN).setParseAction(lambda: [np.inf, False]) \	9✔
26	\| number + Or([RCLOSED, ROPEN])
27
28	interval = left + Suppress(",") + right	9✔
29	interval.setParseAction(lambda t: _Interval(t[1], t[2], t[0], t[3]))	9✔
30
31	expression = Forward()	9✔
32	expression << interval + Optional(Suppress("U") + expression)	9✔
33
34
35	@functools.total_ordering	9✔
36	class _Interval():	9✔
37
38	__slots__ = ['start', 'end', 'start_closed', 'end_closed']	9✔
39
40	def __init__(self,	9✔
41	start=-np.inf,
42	end=np.inf,
43	start_closed=False,
44	end_closed=False):
45	assert start <= end, f"start {start} should not be greater than end {end}."	9✔
46	if start == -np.inf:	9✔
47	assert not start_closed, f"start {start} should not be closed."	9✔
48	if end == np.inf:	9✔
49	assert not end_closed, f"end {end} should not be closed."	9✔
50	self.start = start	9✔
51	self.end = end	9✔
52	self.start_closed = start_closed	9✔
53	self.end_closed = end_closed	9✔
54
55	def __contains__(self, point) -> bool:	9✔
56	if self.start < point < self.end:	9✔
57	return True	9✔
58	if point == self.start and self.start_closed:	9✔
59	return True	×
60	if point == self.end and self.end_closed:	9✔
61	return True	×
62	return False	9✔
63
64	def is_subset_of(self, other: Self) -> bool:	9✔
65	return ((other.start < self.start or other.start == self.start and	9✔
66	(other.start_closed or not self.start_closed))
67	and (self.end < other.end or self.end == other.end and
68	(not self.end_closed or other.end_closed)))
69
70	def intersects(self, other: Self):	9✔
71	if self.start > other.end or self.end < other.start:	9✔
72	return False	×
73	if self.start == other.end and not (self.start_closed	9✔
74	and other.end_closed):
75	return False	×
76	if self.end == other.start and not (self.end_closed	9✔
77	and other.start_closed):
78	return False	9✔
79	return True	9✔
80
81	def intersection(self, other: Self) -> Self:	9✔
82	if not self.intersects(other):	9✔
83	return None	9✔
84	start, start_closed = max((self.start, not self.start_closed),	9✔
85	(other.start, not other.start_closed))
86	start_closed = not start_closed	9✔
87	end, end_closed = min((self.end, self.end_closed),	9✔
88	(other.end, other.end_closed))
89
90	return _Interval(start,	9✔
91	end,
92	start_closed=start_closed,
93	end_closed=end_closed)
94
95	def union(self, other: Self) -> Self:	9✔
96	if not self.intersects(other):	×
97	return None	×
98	start, start_closed = min((self.start, not self.start_closed),	×
99	(other.start, not other.start_closed))
100	start_closed = not start_closed	×
101	end, end_closed = max((self.end, self.end_closed),	×
102	(other.end, other.end_closed))
103	return _Interval(start,	×
104	end,
105	start_closed=start_closed,
106	end_closed=end_closed)
107
108	def empty(self) -> bool:	9✔
109	return self.start > self.end or (	9✔
110	self.start == self.end
111	and not (self.start_closed and self.end_closed))
112
113	def __neg__(self) -> Self:	9✔
114	return _Interval(-self.end, -self.start, self.end_closed,	×
115	self.start_closed)
116
117	def __add__(self, other: int \| float) -> Self:	9✔
118	return _Interval(self.start + other, self.end + other,	×
119	self.start_closed, self.end_closed)
120
121	def __sub__(self, other: int \| float) -> Self:	9✔
122
123	return _Interval(self.start - other, self.end - other,	×
124	self.start_closed, self.end_closed)
125
126	def __mul__(self, other: int \| float) -> Self:	9✔
127	return _Interval(self.start * other, self.end * other,	×
128	self.start_closed, self.end_closed)
129
130	def __truediv__(self, other: int \| float) -> Self:	9✔
131	return _Interval(self.start / other, self.end / other,	×
132	self.start_closed, self.end_closed)
133
134	def __and__(self, other: Self) -> Self:	9✔
135	return self.intersection(other)	×
136
137	def __or__(self, other: Self) -> Self:	9✔
138	return self.union(other)	×
139
140	def __eq__(self, other):	9✔
141	if isinstance(other, (int, float)):	9✔
142	return not self.empty() and self.start == self.end == other	9✔
143	if not isinstance(other, _Interval):	9✔
144	raise TypeError(	×
145	f"Cannot compare interval with {type(other)} {other}.")
146	return (self.start == other.start and self.end == other.end	9✔
147	and self.start_closed == other.start_closed
148	and self.end_closed == other.end_closed)
149
150	def __gt__(self, other):	9✔
151	if isinstance(other, (int, float)):	9✔
152	return self.start > other	9✔
153	if not isinstance(other, _Interval):	9✔
154	raise TypeError(	×
155	f"Cannot compare interval with {type(other)} {other}.")
156	return (self.start, not self.start_closed, self.end,	9✔
157	not self.end_closed) > (other.start, not other.start_closed,
158	other.end, not other.end_closed)
159
160	def __str__(self) -> str:	9✔
161	if self.start == self.end and self.start_closed and self.end_closed:	×
162	return f"{{{self.start}}}"	×
163	elif self.start >= self.end:	×
164	return "{}"	×
165	return f"{'[' if self.start_closed else '('}{self.start}, {self.end}{']' if self.end_closed else ')'}"	×
166
167	def __repr__(self) -> str:	9✔
168	return f"_Interval({self.start}, {self.end}, {self.start_closed}, {self.end_closed})"	×
169
170
171	class Interval():	9✔
172
173	def __init__(self, *args):	9✔
174	if len(args) == 1:	9✔
175	if isinstance(args[0], list):	9✔
176	intervals = args[0]	9✔
177	elif isinstance(args[0], str):	9✔
178	intervals = [	9✔
179	interval.parseString(s)[0] for s in args[0].split('U')
180	]
181	else:
182	raise ValueError(	×
183	'Interval can only be initialized with a list of intervals or a string'
184	)
185	elif len(args) == 2 and isinstance(args[0],	9✔
186	(int, float)) and isinstance(
187	args[1], (int, float)):
188	intervals = [_Interval(*args)]	×
189	elif len(args) == 4 and isinstance(args[0],	9✔
190	(int, float)) and isinstance(
191	args[1], (int, float)):
192	intervals = [_Interval(*args)]	×
193	else:
194	intervals = []	9✔
195	self.intervals: list[_Interval] = sorted(	9✔
196	[interval for interval in intervals if not interval.empty()])
197
198	def empty(self) -> bool:	9✔
199	return len(self.intervals) == 0	9✔
200
201	def full(self) -> bool:	9✔
202	return len(self.intervals) == 1 and self.intervals[	9✔
203	0].start == -np.inf and self.intervals[0].end == np.inf
204
205	def __contains__(self, item: int \| float) -> bool:	9✔
206	if isinstance(item, (int, float)):	9✔
207	i = bisect.bisect_left(self.intervals, item)	9✔
208	if i == 0 or i > len(self.intervals):	9✔
209	return False	9✔
210	return item in self.intervals[i - 1]	9✔
211	else:
212	raise TypeError(f"Cannot check if {item} is in interval.")	×
213
214	def is_subset_of(self, other: Self) -> bool:	9✔
215	if self.empty():	9✔
216	return True	×
217	if other.empty():	9✔
218	return False	×
219	for interval in self.intervals:	9✔
220	if not any(	9✔
221	interval.is_subset_of(other_interval)
222	for other_interval in other.intervals):
223	return False	9✔
224	return True	9✔
225
226	def __eq__(self, other):	9✔
227	if isinstance(other, Interval):	9✔
228	return self.intervals == other.intervals	9✔
229	elif isinstance(other, _Interval):	×
230	return self.intervals == [other]	×
231	else:
232	raise TypeError(	×
233	f"Cannot compare interval with {type(other)} {other}.")
234
235	def __lt__(self, other: Self) -> bool:	9✔
236	return self.is_subset_of(other) and not self == other	×
237
238	def __le__(self, other: Self) -> bool:	9✔
239	return self.is_subset_of(other)	×
240
241	def __gt__(self, other: Self) -> bool:	9✔
242	return other.is_subset_of(self) and not self == other	×
243
244	def __ge__(self, other: Self) -> bool:	9✔
245	return other.is_subset_of(self)	×
246
247	def __neg__(self) -> Self:	9✔
248	return Interval([-interval for interval in self.intervals])	×
249
250	def __add__(self, other: int \| float) -> Self:	9✔
251	if isinstance(other, (int, float)):	×
252	return Interval([interval + other for interval in self.intervals])	×
253	else:
254	raise TypeError(f"Cannot add interval with {type(other)} {other}.")	×
255
256	def __radd__(self, other: int \| float) -> Self:	9✔
257	return self + other	×
258
259	def __sub__(self, other: int \| float) -> Self:	9✔
260	if isinstance(other, (int, float)):	×
261	return Interval([interval - other for interval in self.intervals])	×
262	else:
263	raise TypeError(f"Cannot add interval with {type(other)} {other}.")	×
264
265	def __mul__(self, other: int \| float) -> Self:	9✔
266	if isinstance(other, (int, float)):	×
267	return Interval([interval * other for interval in self.intervals])	×
268	else:
269	raise TypeError(f"Cannot add interval with {type(other)} {other}.")	×
270
271	def __rmul__(self, other: int \| float) -> Self:	9✔
272	return self * other	×
273
274	def __truediv__(self, other: int \| float) -> Self:	9✔
275	if isinstance(other, (int, float)):	×
276	return Interval([interval / other for interval in self.intervals])	×
277	else:
278	raise TypeError(f"Cannot add interval with {type(other)} {other}.")	×
279
280	def __invert__(self):	9✔
281	compl_intervals = []	9✔
282	prev_end = -np.inf	9✔
283	prev_end_closed = False	9✔
284
285	for interval in self.intervals:	9✔
286	compl_intervals.append(	9✔
287	_Interval(prev_end, interval.start, prev_end_closed,
288	not interval.start_closed))
289	prev_end = interval.end	9✔
290	prev_end_closed = not interval.end_closed	9✔
291
292	compl_intervals.append(	9✔
293	_Interval(prev_end, np.inf, prev_end_closed, False))
294	return Interval(compl_intervals)	9✔
295
296	def __and__(self, other):	9✔
297	intersections = []	9✔
298	for i in self.intervals:	9✔
299	for j in other.intervals:	9✔
300	intersection = i.intersection(j)	9✔
301	if intersection is not None:	9✔
302	intersections.append(intersection)	9✔
303	return Interval(intersections)	9✔
304
305	def __or__(self, other):	9✔
306	union = sorted(self.intervals + other.intervals)	9✔
307	merged = [union[0]]	9✔
308	for current in union:	9✔
309	last = merged[-1]	9✔
310	if current.start < last.end or (current.start == last.end and	9✔
311	(last.end_closed
312	or current.start_closed)):
313	if current.end > last.end or (current.end == last.end	9✔
314	and current.end_closed):
315	last.end = current.end	9✔
316	last.end_closed = current.end_closed	9✔
317	else:
318	merged.append(current)	×
319	return Interval(merged)	9✔
320
321	def __repr__(self):	9✔
322	if len(self.intervals) == 0:	×
323	return '{}'	×
324	elif len(self.intervals) == 1:	×
325	return str(self.intervals[0])	×
326	else:
327	return ' U '.join([str(x) for x in self.intervals])	×

feihoo87 / waveforms / 6534953321

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous