Sometimes you’re working on a project and there’s one piece that’s 99% the same as an existing commercial off-the-shelf product, and wouldn’t it be great if you could just integrate that? Today, I’m going to try to do just that, with an off-the-shelf propane tank scale.

One of the things I’ve been meaning to integrate into the hot tub project is a way to gauge propane usage. I don’t fully trust it yet, but ultimately the propane heater is supposed to run on its own, indefinitely, maintaining tub temperature and obviating all but the most strictly necessary human interventions and oversights. Of course, that means it’ll consume propane, and since our house doesn’t have plumbed natural gas to tap, eventually it’ll run the tank empty. This hard fact leads to a couple requirements, in order of strictly necessary (1) to nice-to-have (3):

1. Must detect and alert propane run-out, and disable heating loop pump.
   For one, I want to know when heating has become impossible to avoid the rude surprise of a too-cold tub just when I want to use it. For two, it’d be best not to waste 40W circulating water indefinitely for no purpose.
2. Should gauge propane availability, and calculate whether set point is achievable before run-out and for how long.
   Wouldn’t it be cool if I could set 102°F in the controller interface, and get a readout back that says “not enough propane for that, sorry.” Or even “102 maintainable for 3 days until propane run-out.”
3. Should calculate running cost.
   This one could be either live or after-the-fact calculation based on logged data. Either way, I’d like to know something like “those 3 degrees just cost you $2 in propane” and “maintaining the current temperature burns $4/day.” Live or post-facto, this requires detailed lbs/minute propane usage data.

Option 1: Inferred Metering

One option is to use operating parameters to infer propane usage. For instance, I know the lowest knob setting on the propane heater uses X lbs in Y minutes for a heating rate of ~4.5°F/hr. My control logic could detect this heating rate, infer this propane usage, and integrate accordingly. To satisfy requirement 1 above, all the logic would really need to do is detect “I’ve been ‘heating’ for 5 minutes, but the calculated heat rate is 0, so I must be out.” Indeed, I plan on implementing this logic regardless.

Option 2: Direct Measurement

Of course, implementing option 1 beyond just runout detection seems 1) error prone and 2) like a lot of manual work up front, burning for X minutes and manually measuring propane usage. A more accurate option would be to measure tank weight directly and simply use that data. Luckily, off the shelf solutions for this exist, so it’s “just” a question of integrating one.

Target 1: Flame King

The very first Amazon search result for “propane scale” happens to be both nearly perfect for our use case, and one of the cheapest options: the Flame King smart bluetooth propane tank scale for $20. This is a battery powered bluetooth device that directly measures tank weight. The Raspberry Pi has bluetooth! This is perfect! $20 buys us the following possible strategies

1. Reverse engineer the BLE interface and use the device as-is, with some code on the Pi to poll the tank weight.
2. Replace the firmware on the device to do this, if the protocol can’t be reversed
3. Put in a new controller of our own design to read the sensor directly, then get that data to the Pi wired or wirelessly as we please.

Step 1: Teardown

The first step is to figure out what we’re working with.

As I suspected while shopping, this device uses 3 half-bridge load cells to measure tank weight. If all we got out of this is 3 load cells of sufficient weight handling and a plastic part to hold them correctly, that would probably be worth the $20. Even a fully custom hanging setup would probably be about as expensive, and less convenient. Looking at the controller itself, it uses a FSC-BT647A BLE module from Feasycom, and an STM-8 8-bit microcontroller for the brains. It looks like they’re using a jellybean op amp circuit to average and amplify the 3 load cells electrically, driving a single ADC channel. This would, being analog, be most of the work of replacing the controller entirely.

Step 2: BLE Analysis

So without modification at all, this device does on paper exactly what we need it to. But can we interface it? I started with the same techniques from my article Reverse Engineering Cheap BLE Devices and had a look at the BLE services and characteristics in LightBlue on my iPhone.

There’s one service with a single characteristic that’s “write” – I’m guessing this is for control or firmware updating. There’s one other service with a single characteristic that’s “read notify” which seems like just what we’re after.

If we open this characteristic and click “listen,” sure enough, there’s a 6-byte word that repeatedly updates, changing value with changing weight. I noticed a couple things right off. The full word seems to be divided up into 4 regions:

Region 1 (AA01) never changes. I’m not sure if it’s a sanity check, or a firmware identifier, or what, but it’s always the same. Similarly, region 3 remains pretty constant. I did notice that, when I run the scale from a bench power supply instead of batteries, this section changes with voltage. As it happens right now, the scale has brand new batteries and 0x64 = 0d100, so this sure appears to be the battery level in decimal percent.

Regions 2 and 4 are more interesting. Not only do these both change with applied weight, they don’t remain constant. In the screenshot above, you can see that they adopt 2 values, despite me not changing the weight on the scale:

Step 3: Collecting More Data

At this point, I decided to collect a little more data than I could manually screen-scrape, so I put together a python script in similar fashion to last time. Last time I did this, I used pygatt on Windows with a BlueGiga BLE112. This time, my main machine is back to being a Mac and I want the code to eventually run on the Pi itself, so I opted instead to prototype using Adafruit’s BluefruitLE library as the path of least resistance. Of course, being Adafruit, they have some helpful howto documentation, though unfortunately it doesn’t really address this use case directly. Luckily, PunchThrough, makers of the app we used above, have a slightly more pertinent tutorial that does pretty close to what we need. This is the code I ended up with (Github Gist). Side note: While searching for a cross-platform BLE library, I came across Bleak. It has good-looking cross platform support and even better-looking documentation. If I hadn’t come across the PunchThrough tutorial for the Adafruit library, I’d probably have tried Bleak instead. Maybe in the future.

# Adapted from tutorial code at https://punchthrough.com/bluetooth-low-energy-peripheral-testing/

import time
import uuid
import random
import Adafruit_BluefruitLE

DEVICE_NAME = "Gas Monitor"

# Define service and characteristic UUIDs used by the peripheral.
SERVICE_UUID = uuid.UUID('0000FFE0-0000-1000-8000-00805F9B34FB')
#TX_CHAR_UUID = uuid.UUID('00002222-0000-1000-8000-00805F9B34FB')
RX_CHAR_UUID = uuid.UUID('0000FFE4-0000-1000-8000-00805F9B34FB')

# Get the BLE provider for the current platform.
ble = Adafruit_BluefruitLE.get_provider()


def scan_for_peripheral(adapter):
    """Scan for BLE peripheral and return device if found"""
    print('  Searching for device...')
    try:
        adapter.start_scan()
        # Scan for the peripheral (will time out after 60 seconds
        # but you can specify an optional timeout_sec parameter to change it).
        device = ble.find_device(name=DEVICE_NAME)
        if device is None:
            raise RuntimeError('Failed to find device!')
        return device
    finally:
        # Make sure scanning is stopped before exiting.
        adapter.stop_scan()


def sleep_random(min_ms=1, max_ms=1000):
    """Add a random sleep interval between 1ms to 1000ms"""
    duration_sec = random.randrange(min_ms, max_ms)/1000
    print('   Sleeping for ' + str(duration_sec) + 'sec')
    time.sleep(duration_sec)


def main():
    """Main loop to process BLE events"""
    test_iteration = 0
    echo_mismatch_count = 0
    misc_error_count = 0

    # Clear any cached data because both BlueZ and CoreBluetooth have issues with
    # caching data and it going stale.
    ble.clear_cached_data()

    # Get the first available BLE network adapter and make sure it's powered on.
    adapter = ble.get_default_adapter()
    try:
        adapter.power_on()
        print('Using adapter: {0}'.format(adapter.name))

        # This loop contains the main logic for testing the BLE peripheral.
        # We scan and connect to the peripheral, discover services,
        # read/write to characteristics, and keep track of errors.
        # This test repeats 10 times.
        while test_iteration < 10:
            connected_to_peripheral = False

            while not connected_to_peripheral:
                try:
                    peripheral = scan_for_peripheral(adapter)
                    peripheral.connect(timeout_sec=10)
                    connected_to_peripheral = True
                    test_iteration += 1
                    print('-- Test iteration #{} --'.format(test_iteration))
                except BaseException as e:
                    print("Connection failed: " + str(e))
                    time.sleep(1)
                    print("Retrying...")

            try:
                print('  Discovering services and characteristics...')
                peripheral.discover([SERVICE_UUID], [RX_CHAR_UUID])

                # Find the service and its characteristics
                service = peripheral.find_service(SERVICE_UUID)
                #tx = service.find_characteristic(TX_CHAR_UUID)
                rx = service.find_characteristic(RX_CHAR_UUID)

                # Randomize the intervals between different operations
                # to simulate user-triggered BLE actions.
                # sleep_random(1, 1000)

                # Write random value to characteristic.
                # write_val = bytearray([random.randint(1, 255)])
                # print('  Writing ' + str(write_val) + ' to the write char')
                # tx.write_value(write_val)

                # sleep_random(1, 1000)
                time.sleep(1)
                
                # Function to receive RX characteristic changes.  Note that this will
                # be called on a different thread so be careful to make sure state that
                # the function changes is thread safe.  Use queue or other thread-safe
                # primitives to send data to other threads.
                def received(data):
                    print('Received: {0}'.format(data.hex()))
                    firstSection = int(data.hex()[0:4],16)
                    secondSection = int(data.hex()[4:8],16)
                    thirdSection = int(data.hex()[8:10],16)
                    fourthSection = int(data.hex()[10:],16)
                    print('{0} {1} {2} {3}'.format(firstSection, secondSection, thirdSection, fourthSection))
                    
                rx.start_notify(received)

                #read_val = rx.read_value()
                #print('  Read ' + str(read_val) + ' from the read char')
                # if write_val != read_val:
                #     echo_mismatch_count = echo_mismatch_count + 1
                #     print('  Read value does not match value written')

                time.sleep(60)
                peripheral.disconnect()
                # sleep_random(1, 1000)
            except BaseException as e:
                misc_error_count = misc_error_count + 1
                print('Unexpected error: ' + str(e))
                print('Current error count: ' + str(misc_error_count))
                time.sleep(1)
                print('Retrying...')

    finally:
        # Disconnect device on exit.
        peripheral.disconnect
        print('\nConnection count: ' + str(test_iteration))
        print('Echo mismatch count: ' + str(echo_mismatch_count))
        print('Misc error count: ' + str(misc_error_count))
 
 
# Initialize the BLE system.  MUST be called before other BLE calls!
ble.initialize()

# Start the mainloop to process BLE events, and run the provided function in
# a background thread.  When the provided main function stops running, returns
# an integer status code, or throws an error the program will exit.
ble.run_mainloop_with(main)

This code finds the scale and subscribes to notifications on the “weight” characteristic we identified earlier, then prints out the data as it’s received.

Step 4: Decoding the Data

I loaded up the scale with a few different test weights and collected the data below, noting that every loaded weight produced two output words that seemed to alternate at random, while 0 weight only produced the one word at the top of the table. The hex has been transliterated to decimal format.

This actually contradicts my screenshot above showing two different words, though honestly I may have put some weight on the scale to illustrate the data changing – I don’t recall now. It also raises an interesting point: I converted the data above to decimal, but that might hide patterns from view. Looking at the screenshot data above, Region 2 ends in 08 in both words. I wonder if there’s a similar pattern here that I’ve accidentally masked by only printing decimal in the python output? I went back and added hex representation to the table above:

Sure enough, the lower byte of R1 stays the same regardless of weight. I originally thought the data was deliberately obfuscated in some way, but maybe I simply got my regions wrong. It’s comforting to see a byte that increases linearly with weight, but that leaves me wondering why R4 and the first byte of R1 also change. Clearly, there’s more data to collect and analyze!

To Be Continued.