I need to read a temperature reading from a DS18B20 sensor using Raspberry Pi 3 and Python. The problem is the refresh rate of the sensor (~1 sec) I need to read from sys/bus/w1/devices/28-041670f43bff/w1_slave and use the integer i get to display a temperature on a 7 segment display connected directly to my GPIOs (not using any hardware multiplexing - i2c....etc)

In order to display a two digit temperature, I need to turn on and off the digits really fast (faster than the sensor refreshes)

This is the small piece of code used to get the integer temperature:

def temperature():
    with open ("/sys/bus/w1/devices/28-041670f43bff/w1_slave") as q:
        r=q.read()
        temp=r[69:71]
        t=int (temp)
        return t

But i need to call this function many times per second in order to get a good display on the 7 segment display.

This is how i thought of doing it:

#the temperature() function returns a two digit int
while True:
    GPIO.output(31,0)
    GPIO.output(temp[temperature()/10], 1)  # temp is a dictionary used to know which segments to light up to show numbers
    time.sleep(0.0005)
    GPIO.output(31,1)
    GPIO.output(37,0)
    GPIO.output(temp[temperature()%10], 1)
    time.sleep(0.0005)
    GPIO.output(37,1)

But this code just makes one digit light up, wait ~1sec, light up the other digit, wait ~1sec.....and so on.

Any ideas of how to do this are very appreciated.

2

There are 2 answers

7
Kyler Laird On BEST ANSWER

I'd throw the display routine into its own thread so that you don't have to think about it in your main loop. The code below should demonstrate this concept. Set "testing" to False to see if it works with your hardware.

#!/usr/bin/python

import time
import threading
import Queue
import random

# Set this to False to read the temperature from a real sensor and display it on a 7-digit display.
testing = True

def temperature_read(q):
    # Read the temperature at one second intervals.
    while True:
        if testing:
            r = '-' * 69 + '%02d' % (random.randrange(100)) + 'blahblah' * 4
        else:
            r = open('/sys/bus/w1/devices/28-041670f43bff/w1_slave', 'r').read()

        print r

        # The temperature is represented as two digits in a long string.
        # Push the digits into the queue as a tuple of integers (one per digit).
        q.put((int(r[69]), int(r[70])))

        # Wait for next reading.
        # (Will w1_slave block until the next reading?  If so, this could be eliminated.)
        time.sleep(1.0)

def temperature_display(q):
    # Display the temperature.

    # Temperature is two digits, stored separately (high/low) for more efficient handling.
    temperature_h = temperature_l = 0


    while True:
        # Is there a new temperature reading waiting for us?
        if not q.empty():
            temperature = q.get()

            # If it's None, we're done.
            if temperature is None:
                break   

            # Load the two digits (high and low) representing the temperature.
            (temperature_h, temperature_l) = temperature

        if testing:
            print 'displayH', temperature_h
            time.sleep(0.05)
            print 'displayL', temperature_l
            time.sleep(0.05)

        else:
            GPIO.output(31,0)
            GPIO.output(temperature_h, 1)  # temp is a dictionary used to know which segments to light up to show numbers
            time.sleep(0.0005)
            GPIO.output(31,1)
            GPIO.output(37,0)
            GPIO.output(temperature_l, 1)
            time.sleep(0.0005)
            GPIO.output(37,1)

    # Clean up here.  Turn off all pins?   




# Make a queue to communicate with the display thread.
temperature_queue = Queue.Queue()

# Run the display in a separate thread.
temperature_display_thread = threading.Thread(target=temperature_display, args=(temperature_queue,))
temperature_display_thread.start()

# Run the reader.
try:
        temperature_read(temperature_queue)
except:
        # An uncaught exception happened.  (It could be a keyboard interrupt.)
        None

# Tell the display thread to stop.
temperature_queue.put(None)

# Wait for the thread to end.
temperature_display_thread.join()

To support another reading (transmission), I just put it in the read loop rather than adding another thread for it. I changed the queue so that you could easily move it to another thread but I suspect you'll add more inputs so this is probably a reasonable way to do it unless the read frequency of one needs to be much different. (Even then, you could do things with counters in the loop.)

#!/usr/bin/python

import time
import threading
import Queue
import random

# Set this to False to read the temperature from a real sensor and display it on a 7-digit display.
testing = True

def observe(q):
    while True:

        # Make a temperature reading.
        if testing:
            r = '-' * 69 + '%02d' % (random.randrange(100)) + 'blahblah' * 4
        else:
            r = open('/sys/bus/w1/devices/28-041670f43bff/w1_slave', 'r').read()

        print 'temperature ->', r

        # The temperature is represented as two digits in a long string.
        # Push the digits into the queue as a tuple of integers (one per digit).
        q.put(('temperature', int(r[69]), int(r[70])))

        # Make a transmission reading.
        if testing:
            r = random.randrange(1,6)
        else:
            r = 0   # Put your transmission reading code here.

        print 'transmission ->', r

        q.put(('transmission', r))

        # Wait for next reading.
        # (Will w1_slave block until the next reading?  If so, this could be eliminated.)
        time.sleep(1.0)


def display(q):
    # Display the temperature.

    # Temperature is two digits, stored separately (high/low) for more efficient handling.
    temperature_h = temperature_l = transmission = 0


    while True:
        # Is there a new temperature reading waiting for us?
        if not q.empty():
            reading = q.get()

            # If it's None, we're done.
            if reading is None:
                break   
            elif reading[0] == 'temperature':
                # Load the two digits (high and low) representing the temperature.
                (x, temperature_h, temperature_l) = reading
            elif reading[0] == 'transmission':
                (x, transmission) = reading

        if testing:
            print 'displayH', temperature_h
            time.sleep(0.05)
            print 'displayL', temperature_l
            time.sleep(0.05)
            print 'transmission', transmission
            time.sleep(0.05)

        else:
            GPIO.output(31,0)
            GPIO.output(temperature_h, 1)  # temp is a dictionary used to know which segments to light up to show numbers
            time.sleep(0.0005)
            GPIO.output(31,1)
            GPIO.output(37,0)
            GPIO.output(temperature_l, 1)
            time.sleep(0.0005)
            GPIO.output(37,1)

    # Clean up here.  Turn off all pins?

# Make a queue to communicate with the display thread.
readings_queue = Queue.Queue()

# Run the display in a separate thread.
display_thread = threading.Thread(target=display, args=(readings_queue,))
display_thread.start()

# Observe the inputs.
try:
    observe(readings_queue)
except:
    # An uncaught exception happened.  (It could be a keyboard interrupt.)
    None

# Tell the display thread to stop.
readings_queue.put(None)

# Wait for the thread to end.
display_thread.join()

Here's a version which only reads the temperature every tenth time but reads the transmission every time. I think you'll see how to easily tweak this to meet your needs.

I would make separate threads for each reader but it would complicate the thread management quite a bit.

#!/usr/bin/python

import time
import threading
import Queue
import random

# Set this to False to read the temperature from a real sensor and display it on a 7-digit display.
testing = True

def observe(q):
    count = 0

    while True:
        # Only read the temperature every tenth time.
        if (count % 10 == 0):
            # Make a temperature reading.
            if testing:
                r = '-' * 69 + '%02d' % (random.randrange(100)) + 'blahblah' * 4
            else:
                r = open('/sys/bus/w1/devices/28-041670f43bff/w1_slave', 'r').read()

            print 'temperature ->', r

            # The temperature is represented as two digits in a long string.
            # Push the digits into the queue as a tuple of integers (one per digit).
            q.put(('temperature', int(r[69]), int(r[70])))

        # Make a transmission reading.
        if testing:
            r = random.randrange(1,6)
        else:
            r = 0   # Put your transmission reading code here.

        print 'transmission ->', r

        q.put(('transmission', r))

        # Wait for next reading.
        if testing:
            time.sleep(0.5)
        else:
            time.sleep(0.1)

        count += 1

def display(q):
    # Display the temperature.

    # Temperature is two digits, stored separately (high/low) for more efficient handling.
    temperature_h = temperature_l = transmission = 0


    while True:
        # Is there a new temperature reading waiting for us?
        if not q.empty():
            reading = q.get()

            # If it's None, we're done.
            if reading is None:
                break   
            elif reading[0] == 'temperature':
                # Load the two digits (high and low) representing the temperature.
                (x, temperature_h, temperature_l) = reading
            elif reading[0] == 'transmission':
                (x, transmission) = reading

        if testing:
            print 'displayH', temperature_h
            time.sleep(0.05)
            print 'displayL', temperature_l
            time.sleep(0.05)
            print 'transmission', transmission
            time.sleep(0.05)

        else:
            GPIO.output(31,0)
            GPIO.output(temperature_h, 1)  # temp is a dictionary used to know which segments to light up to show numbers
            time.sleep(0.0005)
            GPIO.output(31,1)
            GPIO.output(37,0)
            GPIO.output(temperature_l, 1)
            time.sleep(0.0005)
            GPIO.output(37,1)

    # Clean up here.  Turn off all pins?

# Make a queue to communicate with the display thread.
readings_queue = Queue.Queue()

# Run the display in a separate thread.
display_thread = threading.Thread(target=display, args=(readings_queue,))
display_thread.start()

# Observe the inputs.
try:
    observe(readings_queue)
except:
    # An uncaught exception happened.  (It could be a keyboard interrupt.)
    None

# Tell the display thread to stop.
readings_queue.put(None)

# Wait for the thread to end.
display_thread.join()
5
Chris Larson On

Rather than implement this functionality on your own, you should instead use the libraries out there that address this particular bit of your code inherently. In this case, I'd suggest you use W1ThermSensor. You can find the documentation at:

https://github.com/timofurrer/w1thermsensor

and you can install it using:

pip install w1thermsensor

It does support the DS18B20, and offers an exact analogue to your use case in the README.

From the docs for the package:

from w1thermsensor import W1ThermSensor

sensor = W1ThermSensor()
temperature_in_celsius = sensor.get_temperature()
temperature_in_fahrenheit = sensor.get_temperature(W1ThermSensor.DEGREES_F)
temperature_in_all_units = sensor.get_temperatures([
    W1ThermSensor.DEGREES_C,
    W1ThermSensor.DEGREES_F,
    W1ThermSensor.KELVIN
])

In many cases, particularly for popular hardware devices, you'll find that there are libraries already available to use within python, and that will all you to quickly move on to writing the bits of code unique to your own particular needs.

Note: According to the technical discussion in the following link, if the DS18B20 is set to 12-bit temperature resolution, the temperature conversion will take 750 ms, or 3/4 of a second. If you set the hardware to do 9-bit resolution, the conversion time in hardware is 93.75 ms. I suspect this is the root of your once-per-second issue.

https://www.maximintegrated.com/en/app-notes/index.mvp/id/4377

There is some discussion of this issue in this Question:

https://raspberrypi.stackexchange.com/questions/14278/how-to-change-ds18b20-reading-resolution

See the second Answer, regarding the configDS18B20 utility.

With the resolution set to 9-bit, you may be able to adjust the w1thermsensor RETRY_DELAY_SECONDS / RETRY_ATTEMPTS value combination in the source code and get what you need. It's unclear to me if the retry delay has any affect on the actual polling of the device. It looks like it is there for device finding. Though, as I said, that interval may impact polling a single device. I simply didn't read through the source code enough to see when and where it comes into play.


Happy New Year!