[docs]
class AcquisitionHistory:
def __init__(self):
'''
Initializes the AcquisitionHistory class
'''
self._source_history = []
self._receiver_deployment_history = []
self._receiver_recovery_history = []
[docs]
def mark_receiver_line_deployed(self, seismic_line, line_name, time):
'''
Creates an instance of AcquisitionHistoryItem, passing the given seismic line, line name and time
Appends that instance to the receiver deployment history
'''
item = AcquisitionHistoryItem(seismic_line, line_name, time)
self._receiver_deployment_history.append(item)
[docs]
def mark_receiver_line_recovered(self, seismic_line, line_name, time):
'''
Creates an instance of AcquisitionHistoryItem, passing the given seismic line, line name and time
Appends that instance to the receiver recovery history
'''
item = AcquisitionHistoryItem(seismic_line, line_name, time)
self._receiver_recovery_history.append(item)
[docs]
def mark_source_line_shot(self, seismic_line, line_name, time):
'''
Creates an instance of AcquisitionHistoryItem, passing the given seismic line, line name and time
Appends that instance to the source history
'''
item = AcquisitionHistoryItem(seismic_line, line_name, time)
self._source_history.append(item)
[docs]
def last_lines_completed(self):
'''
Sets the last shot as the last line name of the source history, or None if there is none
Sets the last deployment as the last line name of the receiver deployment history, or None if there is none
Sets the last recovery as the last line name of the receiver recovery history, or None if there is none
Returns the last shot, last deployment and last recovery
'''
try:
last_shot = self._source_history[-1].line_name
except IndexError:
last_shot = 'None'
try:
last_deployed = self._receiver_deployment_history[-1].line_name
except IndexError:
last_deployed = 'None'
try:
last_recovered = self._receiver_recovery_history[-1].line_name
except IndexError:
last_recovered = 'None'
return last_shot, last_deployed, last_recovered
def _status_at(self, time):
'''
Creates a status distionary with:
- 'shot' as a list of all the seismic lines in the source history if they are completed
- 'deployed' as a list of all the seismic lines in the receiver deployment history if they are completed
- 'recovered' as a list of all the seismic lines in the receiver recovery history if they are completed
Returns the status
'''
status = {
'deployed': [x.seismic_line for x in self._receiver_deployment_history if x.is_completed(time)],
'shot': [x.seismic_line for x in self._source_history if x.is_completed(time)],
'recovered': [x.seismic_line for x in self._receiver_recovery_history if x.is_completed(time)],
}
return status
[docs]
def progress_coords_at(self, time):
'''
Gets the status at the given time with the function '_status_at
Creates an empty dictionary of coordinates with categories 'shot', 'deployed' and 'recovered'
Explains logic::
Loops for each item in the status:
- For each seismic line:
- Gets the extremities (coordinates)
- Appends the coordinates to the right category of the coordinates dictionary
Returns the coordinates dictionary
'''
status = self._status_at(time)
coords = {'shot': [], 'deployed': [], 'recovered': []}
for category, lines in status.items():
for line in lines:
a, b = line.extremities
coords[category].append({'x': (a.x, b.x), 'y': (a.y, b.y)})
return coords
[docs]
class AcquisitionHistoryItem:
def __init__(self, seismic_line, line_name, completion_time):
'''
Initializes the AcquisitionHistoryItem class
'''
self._seismic_line = seismic_line
self._line_name = line_name
self._completion_time = completion_time
@property
def line_name(self):
'''
Returns the line line
'''
return self._line_name
@property
def seismic_line(self):
'''
Returns the seismic line
'''
return self._seismic_line
[docs]
def is_completed(self, time):
'''
Chaeks if the given time is lower (before) the completion time
Returns False if it is not (not completed), and True if it is (completed)
'''
if time < self._completion_time:
return False
return True
def __eq__(self, other):
'''
Equal function: Checks if the given object and the current have the same seismic line and the same line name
Checks if the given object and the current have the same completion time
If both are True, it returns True
Otherwise, it returns False
'''
same_line = (self._seismic_line == other._seismic_line and self._line_name == other._line_name)
same_time = (self._completion_time == other._completion_time)
if same_line and same_time:
return True
return False