[docs]
class HistoryEvent:
def __init__(self, operation, start_time):
self._operation = operation
self._start_time = start_time
self._end_time = start_time + self._operation.duration
[docs]
def clip(self, start_time, end_time):
'''
Checks if the given interval its completely outside the event
If not (it's inside), it returns None
Calculates the actual start as the maximum start between real and given
Calculates the relative start as the actual start minus the real start (could be cero)
Calculates the actual end as the minimum end between real and given
Calculates the relative end as the actual end minus the real start
Clips the event calling the 'clip' function from the instanced 'operation', passing it the relative start and relative end
Wraps that clipped operation in a new instance of the class HistoryEvent, starting at the actual start
Returns that instanced clipped event
'''
interval_outside_of_event = ((start_time <= self.start and end_time <= self.start)
or (start_time >= self.end and end_time >= self.end))
if interval_outside_of_event:
return None
actual_start = max(start_time, self.start)
relative_start = actual_start - self.start
relative_end = min(end_time, self.end) - self.start
clipped = HistoryEvent(self._operation.clip(relative_start, relative_end), actual_start)
return clipped
@property
def name(self):
'''
Returns the name of the operation
'''
return self._operation.name
@property
def receiver_lines_deployed(self):
'''
Returns the number of receiver lines deployed in the operation
'''
return self.operation.receiver_lines_deployed
@property
def receiver_lines_recovered(self):
'''
Returns the number of receiver lines recovered in the operation
'''
return self.operation.receiver_lines_recovered
@property
def source_lines_shot(self):
'''
Returns the number of source lines shot in the operation
'''
return self.operation.source_lines_shot
@property
def operation(self):
'''
Returns the operation
'''
return self._operation
@property
def start(self):
'''
Returns the start time of the operation
'''
return self._start_time
@property
def end(self):
'''
Returns the end time of the operation
'''
return self._end_time
@property
def duration(self):
'''
Returns the duration of the operation as: end time - start time
'''
return self._end_time - self._start_time
[docs]
def is_wait(self):
'''
Returns True if the operation is set as wait, and False otherwise
'''
return self.operation.is_wait()
[docs]
def position_at(self, time):
'''
Checks if the given time is inside the history
If it does, it calculates the position at the relative time (given time minus the start time)
Returns that position
'''
if not self._start_time <= time <= self._end_time:
raise ValueError('Time not within start / end interval: {}'.format(time))
position = self._operation.position_at(time - self._start_time)
return position
[docs]
@classmethod
def duration_of_overlapping_events(cls, events):
'''
Creates a list of the start and end time of each event and calls it intervals
Merges the intervals using the function _merge_intervals
Returns a list of the durations of each merged interval
'''
intervals = [(e.start, e.end) for e in events]
merged = cls._merge_intervals(intervals)
duration = sum([(interval[1] - interval[0]) for interval in merged])
return duration
@staticmethod
def _merge_intervals(time_intervals):
'''
Sorts the given intervals (pairs of start and end time) by the lowest first time, i.e., by the lowest start time
Appends the first interval to a 'merged' list
Loops through the rest of the intervals:
- If the current one overlaps with the last merged interval it merges them by updating the end time
- If they don't overlap, adds the new interval
Returns the list of merged intervals
'''
# adapted from https://codereview.stackexchange.com/questions/69242/merging-overlapping-intervals
sorted_by_lower_bound = sorted(time_intervals, key=lambda tup: tup[0])
merged = []
for interval in sorted_by_lower_bound:
if not merged:
merged.append(interval)
continue
prior_interval = merged[-1]
intervals_intersect = (interval[0] <= prior_interval[1])
if intervals_intersect:
upper_bound = max(prior_interval[1], interval[1])
replacing_interval = (prior_interval[0], upper_bound)
merged[-1] = replacing_interval
else:
merged.append(interval)
return merged
def __eq__(self, other):
'''
Equal function: Checks if the start time and operation of the given are the same as the real
Returns True if they are, and False if they are not
'''
are_equal = (self._start_time == other._start_time and self._operation == other._operation)
if are_equal:
return True
return False
def __repr__(self):
'''
Representation function: Returns a string with the start and end time, next to the string of the associatied operation
'''
repr_string = '{:.1f}h ({:.0f}s) -> {:.1f}h ({:.0f}s): {}'.format(self.start / 3600, self.start,
self.end / 3600, self.end,
self._operation.__repr__())
return repr_string