Cato-CircuitPython/BluetoothControl.py at main · aulitech/Cato-CircuitPython · 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
138
139
140
141
142
143
144
145
146
147
148
149

from adafruit_ble import BLERadio

from adafruit_ble.advertising import Advertisement #, AdvertisingFlag, AdvertisingFlags

from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
from adafruit_ble.services.standard.hid import HIDService
from adafruit_ble.services.standard.device_info import DeviceInfoService
from adafruit_ble.services.standard import BatteryService

from adafruit_hid.keyboard import Keyboard
from adafruit_hid.keyboard_layout_us import KeyboardLayoutUS
from adafruit_hid.keycode import Keycode
from adafruit_hid.mouse import Mouse


from StrUUIDService import SUS
from utils import config

import asyncio

import gc

def mem( loc = "" ):
    print(f"Free Memory at {loc}: \n\t{gc.mem_free()}")

class Appearances:
    remote = 0x0180 #0x0180 to 0x01BF
    eyeglasses = 0x01C0 #0x01C0 to 0x01FF
    hid = 0x03C0 # 0x03C0 to 0x03FF
    control_device = 0x04C0 # 0x04C0 to 0x04FF

class BluetoothControl():

    if(config["name"] == ""):
        config["name"] = f"Cato_{config['HW_UID'][-6:]}"

    # config.dump()

    # BLERadio can toggle advertising state
    ble = BLERadio()
    ble.name = config["name"]

    def __init__(self):
        self.hid = HIDService() # manages human interface device
        print("NEW NAME: ",BluetoothControl.ble.name)
        self.device_info = DeviceInfoService(
            manufacturer = "AULITECH",
            # github will manage
            # on commit, how is a piece of your code updated
            # can set up new repo to play with - google "how to do revision control in github"
            software_revision = "v0.0",
            model_number = None,  # our model number can be "Cato 1"
            serial_number = config["HW_UID"], # look on nrf52840 for hardware serial - can ask on Seeed forum - anything unique per board
            firmware_revision = "v0.0",
            hardware_revision = "v0.0",
            service = None
        )

        name = BluetoothControl.ble.name
        self.advertisement = ProvideServicesAdvertisement( self.hid )
        self.advertisement.appearance = Appearances.hid
        self.advertisement.short_name = name

        # self.advertisement.flags.general_discovery = False
        # self.advertisement.flags.limited_discovery = True
        # self.advertisement.flags.general_discovery = AdvertisingFlag(1)
        # mem("BTC, advertisement created")

        self.scan_response = Advertisement(  )
        self.scan_response.short_name = name
        self.scan_response.appearance = Appearances.remote
        # # mem("BTC, scan_response created")


        # HID handles
        self.k = Keyboard(self.hid.devices)
        self.kl = KeyboardLayoutUS(self.k)
        self.mouse = Mouse(self.hid.devices)

        # Battery service. Can inform central of battery level with this.level
        self.battery_service = BatteryService()

        self.ena_adv = asyncio.Event()          # enable advertising
        self.is_connected = asyncio.Event()     # indicates connection
        self.is_disconnected = asyncio.Event()  # indicates disconnection

        self.is_disconnected.set() #board starts without connection

        import microcontroller as mc
        self.tasks = {}
        if( not(mc.nvm[2]) ): # Should always be true, unless specifically wired training routine
            self.tasks = {  # tasks
                #"characteristic_loop"   : asyncio.create_task(SUS.config_loop()),
                "manage_connection"     : asyncio.create_task(self.manage_connection()),
                "monitor_connections"   : asyncio.create_task(self.monitor_connections()),
                "reconnect"             : asyncio.create_task(self.reconnect())
            }


    async def manage_connection(self):
        #print("+ manage_connection")
        while True:
            # First, wait for advertisement enable
            await self.ena_adv.wait()
            print("Bluetooth: Advertising")
            BluetoothControl.ble.start_advertising(self.advertisement)


            # Then, wait for a connection
            await self.is_connected.wait()

            # Finally, stop advertising
            self.ena_adv.clear()
            self.ble.stop_advertising()
            print("Bluetooth: Advertising disabled")

    async def reconnect(self):
        #print("+ reconnect")
        while True:
            await self.is_disconnected.wait()
            self.ena_adv.set()

            await self.is_connected.wait()
            self.ena_adv.clear()

    async def monitor_connections(self):
        #print("+ monitor_connections")
        while True:
            # Check connection
            if self.ble.connected: # When connected
                if not self.is_connected.is_set():
                    print("Bluetooth: Connected")
                    self.is_connected.set()

                if self.is_disconnected.is_set():
                    self.is_disconnected.clear()

            else: # When disconnected

                if not self.is_disconnected.is_set():
                    print("Bluetooth: Connection Lost")
                    self.is_disconnected.set()

                if self.is_connected.is_set():
                    self.is_connected.clear()

            await asyncio.sleep(3)