0

I've recently stumbled upon this very useful code which I got from this link https://github.com/Douglas6/cputemp. I've modified the code for clarity reasons and so "application" class is imported as an external class. I use this code on my raspberry pi zero w as a bluetooth low energy peripheral. What I don't understand is why set_temperature_callback is called indefinitely so that I may create my own method to send readings (e.g. flex sensor readings).

I appreciate any help!

import dbus

from advertisement import Advertisement
from service import Service, Characteristic, Descriptor
from gpiozero import CPUTemperature

#testing code
#------------------------------------------#
from application import Application
import inspect as i
import sys

def FindSource(method):
    sys.stdout.write(i.getsource(method))
#------------------------------------------#

GATT_CHRC_IFACE = "org.bluez.GattCharacteristic1"

#the interval between sending messages, is counted in milliseconds, was 5000
NOTIFY_TIMEOUT = 1000

class ThermometerAdvertisement(Advertisement):
    def __init__(self, index):
        Advertisement.__init__(self, index, "peripheral")
        self.add_local_name("Raspberry (thermometer)")
        self.include_tx_power = True

class ThermometerService(Service):
    #the service we're interested and is displayed in unity
    THERMOMETER_SVC_UUID = "00000001-710e-4a5b-8d75-3e5b444bc3cf"

    def __init__(self, index):
        self.farenheit = True

        Service.__init__(self, index, self.THERMOMETER_SVC_UUID, True)
        self.add_characteristic(TempCharacteristic(self))
        self.add_characteristic(UnitCharacteristic(self))

    def is_farenheit(self):
        return self.farenheit

    def set_farenheit(self, farenheit):
        self.farenheit = farenheit

class TempCharacteristic(Characteristic):
    TEMP_CHARACTERISTIC_UUID = "00000002-710e-4a5b-8d75-3e5b444bc3cf"

    def __init__(self, service):
        self.notifying = False

        Characteristic.__init__(
                self, self.TEMP_CHARACTERISTIC_UUID,
                ["notify", "read"], service)
        self.add_descriptor(TempDescriptor(self))

    def get_temperature(self):
        value = []
        unit = "C"

        # code here repeats
        cpu = CPUTemperature()
        temp = cpu.temperature
        
        if self.service.is_farenheit():
            temp = (temp * 1.8) + 32
            unit = "F"

        strtemp = str(round(temp, 1)) + " " + unit
        for c in strtemp:
            value.append(dbus.Byte(c.encode()))

        return value

    # the entire method will run on a loop  when unity app hits "subscribe"
    def set_temperature_callback(self):
        if self.notifying:
            print("running inside set_temperature_callback")
            value = self.get_temperature()
            
            self.PropertiesChanged(GATT_CHRC_IFACE, {"Value": value}, [])
        return self.notifying

    def StartNotify(self):
        print("Start notify first time")
        
        if self.notifying:
            print("returning from start notify")
            return

        self.notifying = True

        value = self.get_temperature()
        self.PropertiesChanged(GATT_CHRC_IFACE, {"Value": value}, [])
        self.add_timeout(NOTIFY_TIMEOUT, self.set_temperature_callback)

    def StopNotify(self):
        self.notifying = False

    def ReadValue(self, options):
        value = self.get_temperature()

        return value

class TempDescriptor(Descriptor):
    TEMP_DESCRIPTOR_UUID = "2901"
    TEMP_DESCRIPTOR_VALUE = "CPU Temperature test"

    def __init__(self, characteristic):
        Descriptor.__init__(
                self, self.TEMP_DESCRIPTOR_UUID,
                ["read"],
                characteristic)

    def ReadValue(self, options):
        value = []
        desc = self.TEMP_DESCRIPTOR_VALUE

        for c in desc:
            value.append(dbus.Byte(c.encode()))

        return value

class UnitCharacteristic(Characteristic):
    UNIT_CHARACTERISTIC_UUID = "00000003-710e-4a5b-8d75-3e5b444bc3cf"

    def __init__(self, service):
        Characteristic.__init__(
                self, self.UNIT_CHARACTERISTIC_UUID,
                ["read", "write"], service)
        self.add_descriptor(UnitDescriptor(self))

    def WriteValue(self, value, options):
        val = str(value[0]).upper()
        if val == "C":
            self.service.set_farenheit(False)
        elif val == "F":
            self.service.set_farenheit(True)

    def ReadValue(self, options):
        value = []

        if self.service.is_farenheit(): val = "F"
        else: val = "C"
        value.append(dbus.Byte(val.encode()))

        return value

class UnitDescriptor(Descriptor):
    UNIT_DESCRIPTOR_UUID = "2901"
    UNIT_DESCRIPTOR_VALUE = "Temperature Units (F or C)"

    def __init__(self, characteristic):
        Descriptor.__init__(
                self, self.UNIT_DESCRIPTOR_UUID,
                ["read"],
                characteristic)

    def ReadValue(self, options):
        value = []
        desc = self.UNIT_DESCRIPTOR_VALUE

        for c in desc:
            value.append(dbus.Byte(c.encode()))

        return value

app = Application()
app.add_service(ThermometerService(0))
app.register()

adv = ThermometerAdvertisement(0)
adv.register()
6
  • what do you mean by indefinitely? ... do you mean repeatedly?
    – jsotola
    Nov 10 '21 at 21:10
  • yes exactly, with NOTIFY_TIMEOUT as interval, most likely in milliseconds . Nov 10 '21 at 21:14
  • it appears that you understand why it repeats ... so then, what exactly is your question?
    – jsotola
    Nov 10 '21 at 21:23
  • I don't see where the loop is written. As in, I don't see where the for or while are. Nov 10 '21 at 21:27
  • look for the keyword set_temperature_callback in your code ... you'll find a line that probably sets the timer ... i have no way to test
    – jsotola
    Nov 10 '21 at 21:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.