1# Copyright 2021-2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15# -----------------------------------------------------------------------------
16# Imports
17# -----------------------------------------------------------------------------
18import sys
19import time
20import math
21import random
22import struct
23import logging
24import asyncio
25import os
26
27from bumble.core import AdvertisingData
28from bumble.device import Device
29from bumble.transport import open_transport_or_link
30from bumble.profiles.device_information_service import DeviceInformationService
31from bumble.profiles.heart_rate_service import HeartRateService
32from bumble.utils import AsyncRunner
33
34
35# -----------------------------------------------------------------------------
36async def main() -> None:
37    if len(sys.argv) != 3:
38        print('Usage: python heart_rate_server.py <device-config> <transport-spec>')
39        print('example: python heart_rate_server.py device1.json usb:0')
40        return
41
42    async with await open_transport_or_link(sys.argv[2]) as hci_transport:
43        device = Device.from_config_file_with_hci(
44            sys.argv[1], hci_transport.source, hci_transport.sink
45        )
46
47        # Keep track of accumulated expended energy
48        energy_start_time = time.time()
49
50        def reset_energy_expended():
51            nonlocal energy_start_time
52            energy_start_time = time.time()
53
54        # Add a Device Information Service and Heart Rate Service to the GATT sever
55        device_information_service = DeviceInformationService(
56            manufacturer_name='ACME',
57            model_number='HR-102',
58            serial_number='7654321',
59            hardware_revision='1.1.3',
60            software_revision='2.5.6',
61            system_id=(0x123456, 0x8877665544),
62        )
63
64        heart_rate_service = HeartRateService(
65            read_heart_rate_measurement=lambda _: HeartRateService.HeartRateMeasurement(
66                heart_rate=100 + int(50 * math.sin(time.time() * math.pi / 60)),
67                sensor_contact_detected=random.choice((True, False, None)),
68                energy_expended=random.choice(
69                    (int((time.time() - energy_start_time) * 100), None)
70                ),
71                rr_intervals=random.choice(
72                    (
73                        (
74                            random.randint(900, 1100) / 1000,
75                            random.randint(900, 1100) / 1000,
76                        ),
77                        None,
78                    )
79                ),
80            ),
81            body_sensor_location=HeartRateService.BodySensorLocation.WRIST,
82            reset_energy_expended=lambda _: reset_energy_expended(),
83        )
84
85        device.add_services([device_information_service, heart_rate_service])
86
87        # Set the advertising data
88        device.advertising_data = bytes(
89            AdvertisingData(
90                [
91                    (
92                        AdvertisingData.COMPLETE_LOCAL_NAME,
93                        bytes('Bumble Heart', 'utf-8'),
94                    ),
95                    (
96                        AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
97                        bytes(heart_rate_service.uuid),
98                    ),
99                    (AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340)),
100                ]
101            )
102        )
103
104        # Notify subscribers of the current value as soon as they subscribe
105        @heart_rate_service.heart_rate_measurement_characteristic.on('subscription')
106        def on_subscription(connection, notify_enabled, indicate_enabled):
107            if notify_enabled or indicate_enabled:
108                AsyncRunner.spawn(
109                    device.notify_subscriber(
110                        connection,
111                        heart_rate_service.heart_rate_measurement_characteristic,
112                    )
113                )
114
115        # Go!
116        await device.power_on()
117        await device.start_advertising(auto_restart=True)
118
119        # Notify every 3 seconds
120        while True:
121            await asyncio.sleep(3.0)
122            await device.notify_subscribers(
123                heart_rate_service.heart_rate_measurement_characteristic
124            )
125
126
127# -----------------------------------------------------------------------------
128logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
129asyncio.run(main())
130