kiln_controller/kiln/thermo.py

import re
import time
import random
import datetime
import logging
import threading
from collections import deque, namedtuple
from math import isnan

logger = logging.getLogger("thermo")

def temp_to_cone(temp):
    """Convert the current temperature to cone value using linear interpolation"""
    cones = [600,614,635,683,717,747,792,804,838,852,884,894,900,923,955,984,999,1046,1060,1101,1120,1137,1154,1162,1168,1186,1196,1222,1240,1263,1280,1305,1315,1326,1346]
    names = [str(i).replace('-', '0') for i in range(-22,0)] + [str(i) for i in range(1, 14)]
    for i in range(len(cones)-1):
        low, high = cones[i], cones[i+1]
        if low <= temp < high:
            frac = (temp - low) / float(high - low)
            return names[i]+'.%d'%int(frac*10)
    return "13+"

tempsample = namedtuple("tempsample", ['time', 'temp'])

class MAX31850(object):
    def __init__(self, name="3b-000000182b57", smooth_window=4):
        self.device = "/sys/bus/w1/devices/%s/w1_slave"%name
        self.history = deque(maxlen=smooth_window)
        self.last = None

    def _read_temp(self):
        with open(self.device, 'r') as f:
            lines = f.readlines()

        while lines[0].strip()[-3:] != 'YES':
            time.sleep(0.2)
            with open(self.device, 'r') as f:
                lines = f.readlines()

        match = re.match(r'^[0-9a-f\s]{27}t=(\d+)$', lines[1])
        if match is not None:
            return float(match.group(1)) / 1000.0

    def get(self):
        """Blocking call to retrieve latest temperature sample"""
        self.history.append(self._read_temp())
        self.last = time.time()
        return self.temperature

    @property
    def temperature(self):
        if self.last is None or time.time() - self.last > 5:
            return self.get()

        return tempsample(self.last, sum(self.history) / float(len(self.history)))

class Simulate(object):
    def __init__(self, regulator, smooth_window=120):
        self.regulator = regulator
        self.history = deque(maxlen=smooth_window)
        self.last = None

    def _read_temp(self):
        time.sleep(.25)
        return max([self.regulator.output, 0]) * 1000. + 15+random.gauss(0,.2)

    def get(self):
        self.history.append(self._read_temp())
        self.last = time.time()
        return self.temperature

    @property
    def temperature(self):
        if self.last is None or time.time() - self.last > 5:
            return self.get()

        return tempsample(self.last, sum(self.history) / float(len(self.history)))

class Breakout(object):
    def __init__(self, addr, smooth_window=16):
        import breakout
        self.device = breakout.Breakout(addr)
        self.history = deque(maxlen=smooth_window)
        self.last = None

    def get(self):
        time.sleep(.25)
        temp = self.device.temperature
        if not isnan(temp) and len(self.history) < 1 or abs(temp - self.history[-1]) < 15 or (time.time() - self.last) > 5:
            self.last = time.time()
            self.history.append(temp)
        return self.temperature

    @property
    def temperature(self):
        if self.last is None or time.time() - self.last > 5:
            return self.get()

        return tempsample(self.last, sum(self.history) / float(len(self.history)))

class Monitor(threading.Thread):
    def __init__(self, cls=MAX31850, **kwargs):
        self.therm = cls(**kwargs)
        self.running = True
        
        from Adafruit_alphanumeric import AlphaScroller
        self.display = AlphaScroller(interval=.4)
        self.display.start()
        self.display.hide()

    def run(self):
        while self.running:
            _, temp = self.therm.get()

            if temp > 50:
                if not self.display.shown:
                    self.display.show()
                fahr = temp * 9. / 5. + 32.
                text = list('%0.0f'%temp) + ['degree'] + list('C  %0.0f'%fahr)+['degree'] + list("F")
                if 600 <= temp:
                    text += [' ', ' ', 'cone']+list(temp_to_cone(temp))
                self.display.set_text(text, reset=False)
            elif self.display.shown:
                self.display.hide()

    def stop(self):
        self.running = False
        self.display.stop()


if __name__ == "__main__":
    monitor = Monitor()
    monitor.start()