GPPy/frames.py at master · MichTronics/GPPy · GitHub

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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
from PyQt5.QtGui import QTextCursor
from ax253 import Frame, FrameType, Control
import configparser

CONNECTED = False

serverFormat = '<span style="color:#29F500;">{}</span>'
allFormat = '<span style="color:gray;">{}</span>'
uiFormat = '<span style="color:yellow;">{}</span>'
discFormat = '<span style="color:red;">{}</span>'
messageFormat = '<span style="color:white;">{}</span>'

config = configparser.ConfigParser()


''' Frames Handler for AX25 '''
class FramesHandler:
    def __init__(self) -> None:
        # super(FramesHandler, self).__init__()
        # print("FramesHandler_init")

        pass

    def received_frame(self, frame, ui, ki):
        config.read('./config.ini')
        rframes = self.replace_charcters_html(frame)
        frame = rframes
        self.check_ftype_messages(frame, ui, ki)
        # print(self)
        # print(ki)
        # print(frame)


    def replace_charcters_html(self, frame):
        if b'<' in frame or b'>' in frame:
            a = frame.replace(b'<', b'&lt;')
            b = a.replace(b'>', b'&gt;')
            frame = b
        frame_lt_gt_br = frame.replace(b'\r', b'<br>')
        return frame_lt_gt_br

    def disc(ui, ki):
        frame_disc = Frame.ui(
            destination=frameAX25Source,
            source=config['gp3']['mycall'],
            path=[],
            info="",
            control=FrameType.U_DISC.value,
        )
        ui.tEMonitor.append(discFormat.format(str(frame_disc) +  ' ' + str(frame_disc.control.ftype)))
        ki.write(frame_disc)
        CONNECTED = False
        ui.tECh1.append(discFormat.format("Disconnected from " + str(frameAX25Source)))
        ui.pBChan1.setText("1: ------")
        ui.pBDiscCh1.lower()


    def check_ftype_messages(self, frame, ui, ki):
        global CONNECTED
        global frameAX25Source
        frameAX25 = Frame.from_bytes(frame)
        frameAX25Source = frameAX25.source.callsign.decode()
        frameAX25Destination = frameAX25.destination
        frameAX25Control = frameAX25.control.ftype
        frameAX25Info = frameAX25.info
        frameAX25Split = [str(frameAX25).split(':', 1)[0] + ':'] + str(frameAX25).split(':', 1)[1:]

        if frameAX25Control is FrameType.I:
            ui.tEMonitor.append(allFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))
            ui.tEMonitor.append(uiFormat.format(frameAX25Split[1]))
            if CONNECTED is True:
                ui.tECh1.append(messageFormat.format(frameAX25Split[1]))
                frame_rr = Frame.ui(
                    destination=frameAX25Source,
                    source=config['gp3']['mycall'],
                    path=[],
                    info="",
                    control=FrameType.S_RR.value | (1 << 4) | (1 << 5),
                )
                ui.tEMonitor.append(discFormat.format(str(frame_rr) +  ' ' + str(frame_rr.control.ftype)))
                ki.write(frame_rr)
        elif frameAX25Control is FrameType.U_UI:
            ui.tEMonitor.append(allFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))
            ui.tEMonitor.append(uiFormat.format(frameAX25Split[1]))
        elif frameAX25Control is FrameType.U_UA:
            ui.tEMonitor.append(serverFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))
            if config['gp3']['mycall'] in str(frameAX25Destination) and CONNECTED is False:
                CONNECTED = True
                ui.tECh1.append(serverFormat.format("Connected to " + str(frameAX25Source)))
            else:
                CONNECTED = False
        elif frameAX25Control is FrameType.S_RR:
            ui.tEMonitor.append(allFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))
            if CONNECTED is False:
                frame_disc = Frame.ui(
                    destination=frameAX25Source,
                    source=config['gp3']['mycall'],
                    path=[],
                    info="",
                    control=FrameType.U_DISC.value,
                )
                ui.tEMonitor.append(discFormat.format(str(frame_disc) +  ' ' + str(frame_disc.control.ftype)))
                ki.write(frame_disc)
                # CONNECTED = False
            else:
                frame_rr = Frame.ui(
                    destination=frameAX25Source,
                    source=config['gp3']['mycall'],
                    path=[],
                    info="",
                    control=FrameType.S_RR.value | (1 << 4) | (1 << 5),
                )
                ui.tEMonitor.append(discFormat.format(str(frame_rr) +  ' ' + str(frame_rr.control.ftype)))
                ki.write(frame_rr)

        elif frameAX25Control is FrameType.U_DISC:
            if CONNECTED is True:
                ui.tEMonitor.append(discFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))
                frame_disc = Frame.ui(
                    destination=frameAX25Source,
                    source=config['gp3']['mycall'],
                    path=[],
                    info="",
                    control=FrameType.U_UA.value,
                )
                ui.tEMonitor.append(discFormat.format(str(frame_disc) +  ' ' + str(frame_disc.control.ftype)))
                ki.write(frame_disc)
                CONNECTED = False
                ui.tECh1.append(discFormat.format("Disconnected from " + str(frameAX25Source)))
            else:
                ui.tEMonitor.append(allFormat.format(frameAX25Split[0] + " " + str(frameAX25Control)))

        else:
            ui.tEMonitor.append(allFormat.format(str(frameAX25) + " " + str(frameAX25Control)))
        ui.tEMonitor.moveCursor(QTextCursor.End)