Skip to content
Snippets Groups Projects
controllers.py 50.5 KiB
Newer Older
            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's avatar
Mattia Mancini committed
    def compute_ok_rtsm_list(self):
        mode = COMPONENT_TO_MODE[self.component_type]

        rcus_per_polarization = rcus_from_antenna_and_type(self.antenna_id,
                                                           self.component_type)
Mattia Mancini's avatar
Mattia Mancini committed

        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())). \
            values('pk',
                   'observation_id',
                   'start_datetime',
                   'end_datetime').distinct('pk')
Mattia Mancini's avatar
Mattia Mancini committed
        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's avatar
Mattia Mancini committed
            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). \
Mattia Mancini's avatar
Mattia Mancini committed
            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's avatar
Mattia Mancini committed
            result.append(entry)
        return result

        rcus_per_polarization = rcus_from_antenna_and_type(self.antenna_id,
                                                           self.component_type)

        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(
            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())
                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'])
            rcu = item['rcu']
            polarization = rcus_per_polarization[rcu]
            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()

            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)
        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,
Mattia Mancini's avatar
Mattia Mancini committed
            component__component_id=self.antenna_id).order_by('-station_test__start_datetime')
        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)
            error_type = component_error.type
            details = component_error.details
            errors[station_test_pk]['component_errors'][error_type] = dict(details=details,
                                                                           element_errors=dict())
Mattia Mancini's avatar
Mattia Mancini committed

            error_per_error_type = errors[station_test_pk]['component_errors'][error_type]
Mattia Mancini's avatar
Mattia Mancini committed

            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']
    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 = []
        if self.test_type == 'R' or self.test_type == 'A':
Mattia Mancini's avatar
Mattia Mancini committed
            rtsm_errors_list = self.compute_rtsm_errors_list() + self.compute_ok_rtsm_list()

        if self.test_type == 'S' or self.test_type == 'A':
Mattia Mancini's avatar
Mattia Mancini committed
            station_test_list = self.compute_station_tests_error_list() + self.compute_ok_station_test()
        combined = rtsm_errors_list + station_test_list
        combined_and_sorted = sorted(combined, key=lambda test: test['start_date'], reverse=True)
        payload = OrderedDict()
        payload['errors'] = combined_and_sorted
        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())
        return Response(status=status.HTTP_200_OK, data=payload)