PythonPumpConnector/cm.py at main · OpenMinimed/PythonPumpConnector · 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
from bluezero import dbus_tools
from bluezero.central import Central

import threading
import time

from log_manager import LogManager
from uuids import UUID


class CertificateManagement:
    """
    Certificate Management
    """

    def __init__(self, central:Central):
        self.logger = LogManager.get_logger(self.__class__.__name__)
        self.central = central

        self.cm_cp   = None
        self.cm_data = None

        success = self._configure_characteristics()
        assert success == True

    def unsubscribe(self):
        self.cm_data.add_characteristic_cb(None)
        self.cm_cp.add_characteristic_cb(None)
        return

    def send_request(self):
        ## Certificate Management Control Point char

        # Opcode: 0x00 (Get Certificate)
        #self.cm_cp.write_value([0x00])

        # CertificateManagementOpCode
        # GET_CERTIFICATE(0),
        # GET_ENROLLMENT(1),
        # SET_ENROLLMENT(2),
        # RESPONSE(3),
        # GET_CERTIFICATE_AUTHORITY(4),
        # SET_CERTIFICATE_AUTHORITY(5),
        # GET_REGISTRATION_AUTHORITY(6),
        # SET_REGISTRATION_AUTHORITY(7),
        # GET_FIRMWARE_AUTHORITY(8),
        # SET_FIRMWARE_AUTHORITY(9);
        self.cm_cp.write_value([9])


        return

    def _configure_characteristics(self):
        try:
            # CM service, Certificate Management Control Point characteristic
            self.logger.info("Adding characteristic CMCP")
            chrc = self.central.add_characteristic(
                UUID.CM_SERVICE, UUID.CM_CP_CHAR)
            while not chrc.resolve_gatt():
                time.sleep(0.2)
            assert "write"    in dbus_tools.dbus_to_python(chrc.flags)
            assert "indicate" in dbus_tools.dbus_to_python(chrc.flags)
            chrc.add_characteristic_cb(self._cmcp_cb)
            chrc.start_notify()
            self.cm_cp = chrc
        except Exception as e:
            self.logger.error("Failed to add characteristic CMCP")
            self.logger.error(e)
            return False

        try:
            # CM service, Certificate Managment Data characteristic
            self.logger.info("Adding characteristic Data")
            chrc = self.central.add_characteristic(
                UUID.CM_SERVICE, UUID.CM_DATA_CHAR)
            while not chrc.resolve_gatt():
                time.sleep(0.2)
            assert "write-without-response" in dbus_tools.dbus_to_python(chrc.flags)
            assert "notify"                 in dbus_tools.dbus_to_python(chrc.flags)
            chrc.add_characteristic_cb(self._data_cb)
            chrc.start_notify()
            self.cm_data = chrc
        except Exception as e:
            self.logger.error("Failed to add characteristic Data")
            self.logger.error(e)
            return False

        return True

    def _cmcp_cb(self, iface, changed_props, invalidated_props):
        if "Value" in changed_props:
            value = dbus_tools.dbus_to_python(changed_props["Value"])
            self.logger.debug("CMCP indication: " + value.hex())

    def _data_cb(self, iface, changed_props, invalidated_props):
        if "Value" in changed_props:
            value = dbus_tools.dbus_to_python(changed_props["Value"])
            self.logger.debug("Data notification: " + value.hex())