Newer
Older
Mattia Mancini
committed
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
'error_type',
'percentage',
'count',
'observation__samples').filter(
observation__start_datetime__gt=self.from_date,
observation__end_datetime__lt=self.to_date,
observation__station__name=self.station_name,
mode__in=COMPONENT_TO_MODE[self.component_type],
rcu__in=list(rcus_per_polarization.keys())).order_by('-observation__start_datetime')
for item in rtsm_errors:
observation_pk = item['observation__pk']
if observation_pk not in errors.keys():
errors[observation_pk] = dict(test_id=item['observation__observation_id'],
db_id=item['observation__pk'],
start_date=item[
'observation__start_datetime'],
end_date=item[
'observation__end_datetime'],
test_type='R',
component_errors=dict())
rcu = item['rcu']
polarization = rcus_per_polarization[rcu]
if polarization not in errors[observation_pk]:
errors[observation_pk]['component_errors'][polarization] = dict(
rcu=rcu,
Mattia Mancini
committed
)
error_type = item['error_type']
percentage = item['percentage']
count = item['count']
mode = item['mode']
samples = item['observation__samples']
Mattia Mancini
committed
url_to_plot = reverse('rtsm-summary-plot-detail', (item['pk'],), request=self.request)
Mattia Mancini
committed
errors[observation_pk]['component_errors'][polarization]['errors'][error_type] = dict(samples=samples,
percentage=percentage,
count=count,
mode=mode,
url=url_to_plot)
return list(errors.values())
def compute_station_tests_error_list(self):
errors = dict()
rcus_per_polarization = self.rcus_from_antenna_and_type(self.antenna_id,
self.component_type)
component_errors = ComponentError.objects.filter(
station_test__start_datetime__gt=self.from_date,
station_test__end_datetime__lt=self.to_date,
station_test__station__name=self.station_name,
component__type=self.component_type,
component__component_id=self.antenna_id).order_by('-station_test__start_datetime')
Mattia Mancini
committed
for component_error in component_errors:
station_test_pk = component_error.station_test.pk
if station_test_pk not in errors.keys():
errors[station_test_pk] = dict(test_id=station_test_pk,
db_id=station_test_pk,
start_date=component_error.station_test.start_datetime,
end_date=component_error.station_test.end_datetime,
test_type='S',
component_errors=dict())
for rcu_id, polarization in rcus_per_polarization.items():
if polarization not in errors[station_test_pk]:
errors[station_test_pk]['component_errors'][polarization] = dict(
rcu=rcu_id,
errors=dict())
error_type = component_error.type
details = component_error.details
errors_per_error_polarization = errors[station_test_pk]['component_errors'][polarization]['errors']
errors_per_error_polarization[error_type] = dict(details=details, element_errors=dict())
for element in component_error.failing_elements.values('element__element_id', 'details'):
element_id = element['element__element_id']
errors_per_error_polarization[error_type]['element_errors'][element_id] = element['details']
Mattia Mancini
committed
return list(errors.values())
Mattia Mancini
committed
Mattia Mancini
committed
def compute_response(self):
self.from_date = parse_date(self.from_date)
self.to_date = parse_date(self.to_date)
rtsm_errors_list = []
station_test_list = []
Mattia Mancini
committed
Mattia Mancini
committed
if self.test_type == 'R' or self.test_type == 'A':
rtsm_errors_list = self.compute_rtsm_errors_list() + self.compute_ok_rtsm_list()
Mattia Mancini
committed
if self.test_type == 'S' or self.test_type == 'A':
station_test_list = self.compute_station_tests_error_list() + self.compute_ok_station_test()
Mattia Mancini
committed
Mattia Mancini
committed
combined = rtsm_errors_list + station_test_list
combined_and_sorted = sorted(combined, key=lambda test: test['start_date'], reverse=True)
return Response(status=status.HTTP_200_OK, data=dict(errors=combined_and_sorted))