Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # Save as BLE_server.py
- # Modified to send random integers much faster
- # Tony Goodhew 19th June 2023
- # This is quick and dirty. It uses the Environmental Sensing and Temperature UUIDs
- import bluetooth
- import random
- import struct
- import time
- import machine
- import ubinascii
- from ble_advertising import advertising_payload
- from micropython import const
- from machine import Pin
- import random ############################# EXTRA LINE ###########
- _IRQ_CENTRAL_CONNECT = const(1)
- _IRQ_CENTRAL_DISCONNECT = const(2)
- _IRQ_GATTS_INDICATE_DONE = const(20)
- _FLAG_READ = const(0x0002)
- _FLAG_NOTIFY = const(0x0010)
- _FLAG_INDICATE = const(0x0020)
- # org.bluetooth.service.environmental_sensing
- _ENV_SENSE_UUID = bluetooth.UUID(0x181A)
- # org.bluetooth.characteristic.temperature
- _TEMP_CHAR = (
- bluetooth.UUID(0x2A6E),
- _FLAG_READ | _FLAG_NOTIFY | _FLAG_INDICATE,
- )
- _ENV_SENSE_SERVICE = (
- _ENV_SENSE_UUID,
- (_TEMP_CHAR,),
- )
- # org.bluetooth.characteristic.gap.appearance.xml
- _ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
- class BLETemperature:
- def __init__(self, ble, name=""):
- self._sensor_temp = machine.ADC(4)
- self._ble = ble
- self._ble.active(True)
- self._ble.irq(self._irq)
- ((self._handle,),) = self._ble.gatts_register_services((_ENV_SENSE_SERVICE,))
- self._connections = set()
- if len(name) == 0:
- name = 'Pico %s' % ubinascii.hexlify(self._ble.config('mac')[1],':').decode().upper()
- print('Sensor name %s' % name)
- self._payload = advertising_payload(
- name=name, services=[_ENV_SENSE_UUID]
- )
- self._advertise()
- def _irq(self, event, data):
- # Track connections so we can send notifications.
- if event == _IRQ_CENTRAL_CONNECT:
- conn_handle, _, _ = data
- self._connections.add(conn_handle)
- elif event == _IRQ_CENTRAL_DISCONNECT:
- conn_handle, _, _ = data
- self._connections.remove(conn_handle)
- # Start advertising again to allow a new connection.
- self._advertise()
- elif event == _IRQ_GATTS_INDICATE_DONE:
- conn_handle, value_handle, status = data
- def update_temperature(self, notify=False, indicate=False):
- # Write the local value, ready for a central to read.
- data = random.randint(0,9999) # Generate a random number
- # print("write temp %.2f degc" % temp_deg_c);
- print(data,"Current value being sent");
- self._ble.gatts_write(self._handle, struct.pack("<h", data)) ###
- if notify or indicate:
- for conn_handle in self._connections:
- if notify:
- # Notify connected centrals.
- self._ble.gatts_notify(conn_handle, self._handle)
- if indicate:
- # Indicate connected centrals.
- self._ble.gatts_indicate(conn_handle, self._handle)
- def _advertise(self, interval_us=5000): # was 500000
- self._ble.gap_advertise(interval_us, adv_data=self._payload)
- '''
- def _get_temp(self): #################### Modified here ##########
- x = random.randint(0,9999)
- return float(x)
- '''
- def demo():
- ble = bluetooth.BLE()
- temp = BLETemperature(ble)
- counter = 0
- led = Pin('LED', Pin.OUT)
- while True:
- if counter % 2 == 0:
- temp.update_temperature(notify=True, indicate=False)
- led.toggle()
- time.sleep_ms(100)
- counter += 1
- if __name__ == "__main__":
- demo()
- ############ SECOND PROGRAM BELOW ############
- # Save as BLE_reader.py
- # Finds and connects to a BLE temperature sensor but reads integers
- # Modified to receive random integers much faster
- # Tony Goodhew 19th June 2023
- import bluetooth
- import random
- import struct
- import time
- import micropython
- from ble_advertising import decode_services, decode_name
- from micropython import const
- from machine import Pin
- _IRQ_CENTRAL_CONNECT = const(1)
- _IRQ_CENTRAL_DISCONNECT = const(2)
- _IRQ_GATTS_WRITE = const(3)
- _IRQ_GATTS_READ_REQUEST = const(4)
- _IRQ_SCAN_RESULT = const(5)
- _IRQ_SCAN_DONE = const(6)
- _IRQ_PERIPHERAL_CONNECT = const(7)
- _IRQ_PERIPHERAL_DISCONNECT = const(8)
- _IRQ_GATTC_SERVICE_RESULT = const(9)
- _IRQ_GATTC_SERVICE_DONE = const(10)
- _IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
- _IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
- _IRQ_GATTC_DESCRIPTOR_RESULT = const(13)
- _IRQ_GATTC_DESCRIPTOR_DONE = const(14)
- _IRQ_GATTC_READ_RESULT = const(15)
- _IRQ_GATTC_READ_DONE = const(16)
- _IRQ_GATTC_WRITE_DONE = const(17)
- _IRQ_GATTC_NOTIFY = const(18)
- _IRQ_GATTC_INDICATE = const(19)
- _ADV_IND = const(0x00)
- _ADV_DIRECT_IND = const(0x01)
- _ADV_SCAN_IND = const(0x02)
- _ADV_NONCONN_IND = const(0x03)
- # org.bluetooth.service.environmental_sensing
- _ENV_SENSE_UUID = bluetooth.UUID(0x181A)
- # org.bluetooth.characteristic.temperature
- _TEMP_UUID = bluetooth.UUID(0x2A6E)
- _TEMP_CHAR = (
- _TEMP_UUID,
- bluetooth.FLAG_READ | bluetooth.FLAG_NOTIFY,
- )
- _ENV_SENSE_SERVICE = (
- _ENV_SENSE_UUID,
- (_TEMP_CHAR,),
- )
- class BLETemperatureCentral:
- def __init__(self, ble):
- self._ble = ble
- self._ble.active(True)
- self._ble.irq(self._irq)
- self._reset()
- self._led = Pin('LED', Pin.OUT)
- def _reset(self):
- # Cached name and address from a successful scan.
- self._name = None
- self._addr_type = None
- self._addr = None
- # Cached value (if we have one)
- self._value = None
- # Callbacks for completion of various operations.
- # These reset back to None after being invoked.
- self._scan_callback = None
- self._conn_callback = None
- self._read_callback = None
- # Persistent callback for when new data is notified from the device.
- self._notify_callback = None
- # Connected device.
- self._conn_handle = None
- self._start_handle = None
- self._end_handle = None
- self._value_handle = None
- def _irq(self, event, data):
- if event == _IRQ_SCAN_RESULT:
- addr_type, addr, adv_type, rssi, adv_data = data
- if adv_type in (_ADV_IND, _ADV_DIRECT_IND):
- type_list = decode_services(adv_data)
- if _ENV_SENSE_UUID in type_list:
- # Found a potential device, remember it and stop scanning.
- self._addr_type = addr_type
- self._addr = bytes(addr) # Note: addr buffer is owned by caller so need to copy it.
- self._name = decode_name(adv_data) or "?"
- self._ble.gap_scan(None)
- elif event == _IRQ_SCAN_DONE:
- if self._scan_callback:
- if self._addr:
- # Found a device during the scan (and the scan was explicitly stopped).
- self._scan_callback(self._addr_type, self._addr, self._name)
- self._scan_callback = None
- else:
- # Scan timed out.
- self._scan_callback(None, None, None)
- elif event == _IRQ_PERIPHERAL_CONNECT:
- # Connect successful.
- conn_handle, addr_type, addr = data
- if addr_type == self._addr_type and addr == self._addr:
- self._conn_handle = conn_handle
- self._ble.gattc_discover_services(self._conn_handle)
- elif event == _IRQ_PERIPHERAL_DISCONNECT:
- # Disconnect (either initiated by us or the remote end).
- conn_handle, _, _ = data
- if conn_handle == self._conn_handle:
- # If it was initiated by us, it'll already be reset.
- self._reset()
- elif event == _IRQ_GATTC_SERVICE_RESULT:
- # Connected device returned a service.
- conn_handle, start_handle, end_handle, uuid = data
- if conn_handle == self._conn_handle and uuid == _ENV_SENSE_UUID:
- self._start_handle, self._end_handle = start_handle, end_handle
- elif event == _IRQ_GATTC_SERVICE_DONE:
- # Service query complete.
- if self._start_handle and self._end_handle:
- self._ble.gattc_discover_characteristics(
- self._conn_handle, self._start_handle, self._end_handle)
- else:
- print("Failed to find environmental sensing service.")
- elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
- # Connected device returned a characteristic.
- conn_handle, def_handle, value_handle, properties, uuid = data
- if conn_handle == self._conn_handle and uuid == _TEMP_UUID:
- self._value_handle = value_handle
- elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
- # Characteristic query complete.
- if self._value_handle:
- # We've finished connecting and discovering device, fire the connect callback.
- if self._conn_callback:
- self._conn_callback()
- else:
- print("Failed to find temperature characteristic.")
- elif event == _IRQ_GATTC_READ_RESULT:
- # A read completed successfully.
- conn_handle, value_handle, char_data = data
- if conn_handle == self._conn_handle and value_handle == self._value_handle:
- self._update_value(char_data)
- if self._read_callback:
- self._read_callback(self._value)
- self._read_callback = None
- elif event == _IRQ_GATTC_READ_DONE:
- # Read completed (no-op).
- conn_handle, value_handle, status = data
- elif event == _IRQ_GATTC_NOTIFY:
- # The ble_temperature.py demo periodically notifies its value.
- conn_handle, value_handle, notify_data = data
- if conn_handle == self._conn_handle and value_handle == self._value_handle:
- self._update_value(notify_data)
- if self._notify_callback:
- self._notify_callback(self._value)
- # Returns true if we've successfully connected and discovered characteristics.
- def is_connected(self):
- return self._conn_handle is not None and self._value_handle is not None
- # Find a device advertising the environmental sensor service.
- def scan(self, callback=None):
- self._addr_type = None
- self._addr = None
- self._scan_callback = callback
- self._ble.gap_scan(2000, 30000, 30000)
- # Connect to the specified device (otherwise use cached address from a scan).
- def connect(self, addr_type=None, addr=None, callback=None):
- self._addr_type = addr_type or self._addr_type
- self._addr = addr or self._addr
- self._conn_callback = callback
- if self._addr_type is None or self._addr is None:
- return False
- self._ble.gap_connect(self._addr_type, self._addr)
- return True
- # Disconnect from current device.
- def disconnect(self):
- if not self._conn_handle:
- return
- self._ble.gap_disconnect(self._conn_handle)
- self._reset()
- # Issues an (asynchronous) read, will invoke callback with data.
- def read(self, callback):
- if not self.is_connected():
- return
- self._read_callback = callback
- try:
- self._ble.gattc_read(self._conn_handle, self._value_handle)
- except OSError as error:
- print(error)
- # Sets a callback to be invoked when the device notifies us.
- def on_notify(self, callback):
- self._notify_callback = callback
- def _update_value(self, data):
- # Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
- try:
- self._value = struct.unpack("<h", data)[0] ####### Modified
- except OSError as error:
- print(error)
- def value(self):
- return self._value
- def sleep_ms_flash_led(self, flash_count, delay_ms):
- self._led.off()
- while(delay_ms > 0):
- for i in range(flash_count):
- self._led.on()
- time.sleep_ms(100)
- self._led.off()
- time.sleep_ms(100)
- delay_ms -= 200
- time.sleep_ms(10) # was 1000
- delay_ms -= 1000
- def print_temp(result):
- print(result, "Received") ### Modified
- def demo(ble, central):
- not_found = False
- def on_scan(addr_type, addr, name):
- if addr_type is not None:
- print("Found sensor: %s" % name)
- central.connect()
- else:
- nonlocal not_found
- not_found = True
- print("No sensor found.")
- central.scan(callback=on_scan)
- # Wait for connection...
- while not central.is_connected():
- time.sleep_ms(100)
- if not_found:
- return
- print("Connected")
- # Explicitly issue reads
- while central.is_connected():
- central.read(callback=print_temp)
- sleep_ms_flash_led(central, 1, 2) ## was 2000
- print("Disconnected")
- if __name__ == "__main__":
- ble = bluetooth.BLE()
- central = BLETemperatureCentral(ble)
- while(True):
- demo(ble, central)
- sleep_ms_flash_led(central, 1, 10000)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement