Newer
Older
Mattia Mancini
committed
'test_type',
required=False,
Mattia Mancini
committed
location='query',
Mattia Mancini
committed
schema=coreschema.Enum(
['R', 'S', 'A'],
description='select the type of test possible values are (R, RTSM),'
' (S, Station test), (A, all)[DEFAULT=A]',
)
)
]
Mattia Mancini
committed
mode = COMPONENT_TO_MODE[self.component_type]
rcus_per_polarization = rcus_from_antenna_and_type(self.antenna_id,
self.component_type)
good_observation_list = RTSMObservation.objects.filter(errors_summary__mode__in=mode,
start_datetime__gt=self.from_date,
end_datetime__lt=self.to_date,
station__name=self.station_name). \
exclude(errors_summary__rcu__in=list(rcus_per_polarization.keys())). \
'observation_id',
'start_datetime',
'end_datetime').distinct('pk')
result = []
for observation in good_observation_list:
entry = dict(test_id=observation['observation_id'],
db_id=observation['pk'],
start_date=observation['start_datetime'],
end_date=observation['end_datetime'],
test_type='R',
component_errors=dict())
antenna_statuses = latest_status_per_station_and_component_type_antenna_id(
WinCCAntennaStatus.objects,
self.station_name,
self.component_type,
self.antenna_id,
to_date=observation[
'end_datetime'])
Mattia Mancini
committed
entry['status'] = antenna_statuses
result.append(entry)
return result
def compute_ok_station_test(self):
good_station_test = StationTest.objects.filter(start_datetime__gt=self.from_date,
end_datetime__lt=self.to_date,
station__name=self.station_name). \
exclude(component_errors__component__component_id=self.antenna_id). \
values('pk', 'start_datetime', 'end_datetime').order_by('-start_datetime')
result = []
for station_test in good_station_test:
entry = dict(test_id=station_test['pk'],
db_id=station_test['pk'],
start_date=station_test['start_datetime'],
end_date=station_test['end_datetime'],
test_type='S',
component_errors=dict())
antenna_statuses = latest_status_per_station_and_component_type_antenna_id(
WinCCAntennaStatus.objects,
self.station_name,
self.component_type,
self.antenna_id,
to_date=station_test['end_datetime'])
Mattia Mancini
committed
entry['status'] = antenna_statuses
Mattia Mancini
committed
def compute_rtsm_errors_list(self):
errors = dict()
rcus_per_polarization = rcus_from_antenna_and_type(self.antenna_id,
self.component_type)
Mattia Mancini
committed
rtsm_errors = RTSMErrorSummary.objects.values('observation__pk',
'observation__start_datetime',
'observation__end_datetime',
'observation__observation_id',
'observation__station__name',
'pk',
'rcu',
'mode',
'error_type',
'percentage',
'count',
'observation__samples').filter(
Mattia Mancini
committed
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')
Mattia Mancini
committed
Mattia Mancini
committed
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())
Mattia Mancini
committed
antenna_statuses = latest_status_per_station_and_component_type_antenna_id(
WinCCAntennaStatus.objects,
self.station_name,
self.component_type,
self.antenna_id,
to_date=item[
'observation__end_datetime'])
Mattia Mancini
committed
errors[observation_pk]['status'] = antenna_statuses
Mattia Mancini
committed
rcu = item['rcu']
polarization = rcus_per_polarization[rcu]
Mattia Mancini
committed
error_type = item['error_type']
percentage = item['percentage']
count = item['count']
mode = item['mode']
samples = item['observation__samples']
if error_type not in errors[observation_pk]['component_errors']:
errors[observation_pk]['component_errors'][error_type] = dict()
Mattia Mancini
committed
url_to_plot = reverse('rtsm-summary-plot-detail', (item['pk'],), request=self.request)
errors[observation_pk]['component_errors'][error_type][polarization] = dict(
samples=samples,
percentage=percentage,
count=count,
mode=mode,
rcu=rcu,
url=url_to_plot)
Mattia Mancini
committed
return list(errors.values())
def compute_station_tests_error_list(self):
errors = dict()
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())
antenna_statuses = latest_status_per_station_and_component_type_antenna_id(
WinCCAntennaStatus.objects,
self.station_name,
self.component_type,
self.antenna_id,
to_date=component_error.station_test.end_datetime)
Mattia Mancini
committed
errors[station_test_pk]['status'] = antenna_statuses
error_type = component_error.type
details = component_error.details
errors[station_test_pk]['component_errors'][error_type] = dict(details=details,
element_errors=dict())
error_per_error_type = errors[station_test_pk]['component_errors'][error_type]
for element in component_error.failing_elements.values('element__element_id',
'details'):
element_id = element['element__element_id']
error_per_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)
Mattia Mancini
committed
payload = OrderedDict()
payload['errors'] = combined_and_sorted
Mattia Mancini
committed
payload['current_status'] = latest_status_per_station_and_component_type_antenna_id(
WinCCAntennaStatus.objects,
self.station_name,
self.component_type,
self.antenna_id,
to_date=datetime.datetime.now())
Mattia Mancini
committed
return Response(status=status.HTTP_200_OK, data=payload)