Skip to content
Snippets Groups Projects
Select Git revision
  • eive/master
  • master default protected
  • fsfw/master
  • v1.1
  • v1.0
5 results
Blame Permalink
Forked from an inaccessible project.
obsw_tm_service_1.py 4.18 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101 






# -*- coding: utf-8 -*-
"""
Program: obsw_tm_service_1.py
Date: 30.12.2019
Description: Deserialize Pus Verification TM
Author: R. Mueller
"""
import struct
from typing import Dict

from tm.obsw_pus_tm_base import PusTelemetry, TmDictionaryKeys
from tm.obsw_pus_tm_creator import PusTelemetryCreator

pusPacketInfoService1T = Dict[TmDictionaryKeys, any]


class Service1TM(PusTelemetry):
    def __init__(self, byte_array: bytearray):
        super().__init__(byte_array)
        self.has_tc_error_code = False
        self.is_step_reply = False
        # Failure Reports with error code
        self.err_code = 0
        self.step_number = 0
        self.tcPacketId = self._tm_data[0] << 8 | self._tm_data[1]
        self.tcSSC = ((self._tm_data[2] & 0x3F) << 8) | self._tm_data[3]
        self.specify_packet_info("Success Verification")
        if self.get_subservice() % 2 == 0:
            self.specify_packet_info("Failure Verficiation")
            self.has_tc_error_code = True
            if self.get_subservice() == 6:
                self.is_step_reply = True
                self.append_packet_info(" : Step Failure")
                self.step_number = struct.unpack('>B', self._tm_data[4:5])[0]
                self.err_code = struct.unpack('>H', self._tm_data[5:7])[0]
                self.errorParam1 = struct.unpack('>I', self._tm_data[7:11])[0]
                self.errorParam2 = struct.unpack('>I', self._tm_data[11:15])[0]
            else:
                self.print_source_data()
                self.err_code = struct.unpack('>H', self._tm_data[4:6])[0]
                self.errorParam1 = struct.unpack('>I', self._tm_data[6:10])[0]
                self.errorParam2 = struct.unpack('>I', self._tm_data[10:14])[0]

        elif self.get_subservice() == 5:
            self.is_step_reply = True
            self.append_packet_info(" : Step Success")
            self.step_number = struct.unpack('>B', self._tm_data[4:5])[0]

    def append_telemetry_content(self, array):
        super().append_telemetry_content(array)
        array.append(str(hex(self.tcPacketId)))
        array.append(str(self.tcSSC))
        if self.has_tc_error_code:
            if self.is_step_reply:
                array.append(str(self.step_number))
            array.append(str(hex(self.err_code)))
            array.append(str(hex(self.errorParam1)) + ", " + str(self.errorParam1))
            array.append(str(hex(self.errorParam2)) + ", " + str(self.errorParam2))
        elif self.is_step_reply:
            array.append(str(self.step_number))

    def append_telemetry_column_headers(self, array):
        super().append_telemetry_column_headers(array)
        array.append("TC Packet ID")
        array.append("TC SSC")
        if self.has_tc_error_code:
            if self.is_step_reply:
                array.append("Step Number")
            array.append("Return Value")
            array.append("Error Param 1")
            array.append("Error Param 2")
        elif self.is_step_reply:
            array.append("Step Number")

    def pack_tm_information(self) -> pusPacketInfoService1T:
        tm_information = super().pack_tm_information()
        add_information = {
            TmDictionaryKeys.TC_PACKET_ID: self.tcPacketId,
            TmDictionaryKeys.TC_SSC: self.tcSSC,
        }
        tm_information.update(add_information)
        if self.has_tc_error_code:
            tm_information.update({TmDictionaryKeys.ERROR_CODE: self.err_code})
        if self.is_step_reply:
            tm_information.update({TmDictionaryKeys.STEP_NUMBER: self.step_number})
        return tm_information


class Service1TmPacked(PusTelemetryCreator):
    def __init__(self, subservice: int, ssc: int=0, tc_packed_id: int=0, tc_ssc: int=0):
        source_data = bytearray()
        source_data.append((tc_packed_id & 0xFF00) >> 8)
        source_data.append(tc_packed_id & 0xFF)
        tc_psc =  (tc_ssc & 0x3FFF) | (0b11 << 16)
        source_data.append((tc_psc & 0xFF00) >> 8)
        source_data.append(tc_psc & 0xFF)
        super().__init__(service=1, subservice=subservice, ssc=ssc, source_data=source_data)

    def pack(self) -> bytearray:
        return super().pack()