-
Notifications
You must be signed in to change notification settings - Fork 19
/
utils.py
132 lines (99 loc) · 3.67 KB
/
utils.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
import functools as ft
from itertools import groupby
from operator import itemgetter
from typing import List, Callable, Any, Generator
def forward_merge(merge_list: List, condition: Callable[[Any], bool]) -> List:
"""
Helper function to merge items in a list by adding them to the next eligible element.
This merge moves left to right.
For example, given the list:
[1, 2, 3, 4, 5]
And the condition, x < 3, the function yields:
>>> forward_merge([1,2,3,4,5], lambda x: x < 3)
>>> [6, 4, 5]
:param merge_list: the list to merge
:param condition: the merge condition
:return: a list of the merged items
"""
items = []
def _flatten(ml):
return ft.reduce(lambda acc, x: acc + x, ml)
merge_items = []
merge_index = None
for i, item in enumerate(merge_list):
if condition(item):
merge_items.append(item)
elif merge_items:
# we found a large item and have short items to merge
merge_items.append(item)
items.append(_flatten(merge_items))
merge_items = []
merge_index = i
else:
items.append(item)
if merge_items and not merge_index:
# we got to the end but still have merge items;
items.append(_flatten(merge_items))
else:
for item in merge_items:
items.append(item)
return items
def reverse_merge(merge_list: List, condition: Callable[[Any], bool]) -> List:
"""
Helper function to merge items in a list by adding them to the next eligible element.
This merge moves right to left.
For example, given the list:
[1, 2, 3, 4, 5]
And the condition, x < 3, the function yields:
>>> list(reverse_merge([1,2,3,4,5], lambda x: x < 3))
>>> [3, 3, 4, 5]
:param merge_list: the list to merge
:param condition: the merge condition
:return: a list of the merged items
"""
items = []
def _flatten(ml):
return ft.reduce(lambda acc, x: x + acc, ml)
merge_items = []
merge_index = None
for i in reversed(range(len(merge_list))):
item = merge_list[i]
if condition(item):
merge_items.append(item)
elif merge_items:
# we found a large item and have short items to merge
merge_items.append(item)
items.append(_flatten(merge_items))
merge_items = []
merge_index = i
else:
items.append(item)
if merge_items and not merge_index:
# we got to the end but still have merge items;
items.append(_flatten(merge_items))
else:
for item in merge_items:
items.append(item)
return list(reversed(items))
def merge(merge_list: List, condition: Callable[[Any], bool]) -> List:
"""
combines the forward and reverse merges to catch edge cases at the tail ends of the list
:param merge_list: the list to merge
:param condition: the merge condition
:return: a list of the merged items
"""
f_merge = forward_merge(merge_list, condition)
if any(map(condition, f_merge)):
return reverse_merge(f_merge, condition)
else:
return f_merge
def compress(cutting_points: List) -> Generator:
"""
compress a list of cutting points if they happen to be directly adjacent to another
:param cutting_points: the list of cutting points to compress
:return: the compressed list of cutting points
"""
sorted_cuts = sorted(cutting_points, key=lambda c: c.trace_index)
for k, g in groupby(enumerate(sorted_cuts), lambda x: x[0] - x[1].trace_index):
all_cps = list(map(itemgetter(1), g))
yield all_cps[int(len(all_cps) / 2)]