-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathex.py
33 lines (25 loc) · 1.24 KB
/
ex.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
from functools import reduce
from operator import concat
# ===================================================
# utils
# ===================================================
def select_keys(d: dict, ks: list) -> dict:
return {k: d[k] for k in ks if k in d}
# ===================================================
# API
# ===================================================
def produce_api_calls(intervals: list) -> list:
intervals = [select_keys(x, ['vessel', 'start', 'end']) for x in intervals]
intervals = sorted(intervals, key=lambda x: (x['vessel'], x['start']))
v_to_intervals = {}
for interval in intervals:
vessel, start2, end2 = interval.get('vessel'), interval.get('start'), interval.get('end')
cur_intervals = v_to_intervals.get(vessel, [])
if not cur_intervals or cur_intervals[-1].get('end') < start2:
# empty or no_overlap (end1<start2): add interval
cur_intervals.append(interval)
v_to_intervals[vessel] = cur_intervals
else:
# overlaps (start2 <= end1): update cur interval with [start1, max(end1, end2)]
v_to_intervals[vessel][-1]['end'] = max(cur_intervals[-1].get('end'), end2)
return list(reduce(concat, v_to_intervals.values()))