This isn’t the most involved project but I might as well document it. I’ve been trying to automate some of the more boring tasks of running the evaporator, I’ve got some nice stainless steel float valves to regulate the sap going in now for example. One of the things that kept requiring constant attention is the air intake to adjust the strength of the fire. I’d have to sit with a foot on it to be able to regulate it almost constantly, to make sure the fire wasn’t burning too hard or too weak. And so naturally I thought I could do something with a Pi.
This proved quite successful even with very loose wiring and fastening just to see how it would work.
All of a sudden I barely need to pay attention to the fire’s strength, with a few refinements I won’t have to at all.
The circuit is quite simple:
Wifi barely reaches the sugarhouse so I made sure this could run independent of connectivity. Which involves coding threads on a Pi Pico, which is supported but not as one would expect.
main.pyimport machine import time import network import socket from max6675 import MAX6675 import _thread html = """{\"evaporator_temperature\":<TEMPERATURE>}""" # LED led = machine.Pin( "LED", machine.Pin.OUT ) # temperature sck = machine.Pin( 2, machine.Pin.OUT ) cs = machine.Pin( 3, machine.Pin.OUT ) so = machine.Pin( 4, machine.Pin.IN ) sensor = MAX6675( sck, cs , so ) temperature_min = 25 temperature_max = 30 temperature = -1337.0 # servo servo = machine.PWM( machine.Pin(0) ) servo.freq( 50 ) servo_min = 1000 servo_max = 8000 servo_at = 0 def temp_to_servo( temp ): if (temperature_max - temperature_min)==0: # right in the middle return int( (servo_max-servo_min)/2 ) result = (temp - temperature_min) * (servo_max - servo_min) / (temperature_max - temperature_min) + servo_min if result>servo_max: result = servo_max if result<servo_min: result = servo_min return int( result ) def blink_number( number ): number = str( int(number) ) for char in number: for i in range(int(char)): led.value( 1 ) time.sleep( 0.2 ) led.value( 0 ) time.sleep( 0.2 ) time.sleep( 0.3 ) led.value( 1 ) time.sleep( 0.1 ) led.value( 0 ) time.sleep( 0.3 ) keep_going = False def servo_thread(): global temperature, servo_at, servo while keep_going: time.sleep( 5 ) print( "# measuring average temperature over 1 seconds..." ) temperature_total = 0.0 for i in range(10): temperature_total += sensor.read() time.sleep( 0.1 ) temperature = temperature_total / 10 blink_number( temperature ) print( "# " + str(temperature) ) new_servo_position = temp_to_servo( temperature ) print( "# new_servo_position: " + str(new_servo_position) ) step = 1 if new_servo_position<servo_at: step = -1 for i in range(servo_at, new_servo_position, step): time.sleep( 0.001 ) servo_at = i servo.duty_u16( i ) print( "# servo tread finishing" ) # main thread servo.duty_u16( servo_min ) for i in range(servo_min, servo_max): time.sleep( 0.001 ) servo_at = i servo.duty_u16( i ) try: ssid = "<wifi_ssid>" password = "<wifi_password>" wlan = network.WLAN( network.STA_IF ) wlan.active( True ) wlan.connect( ssid, password ) # wait for connect or fail max_wait = 20 while max_wait>0: if wlan.status() < 0 or wlan.status() >= 3: break max_wait -= 1 print( "> waiting for connection..." ) time.sleep( 1 ) # handle connection error if wlan.status()!=3: print( "> network connection failed, will launch servo thread in 1 minute" ) time.sleep( 60 ) print( "> launching" ) keep_going = True _thread.start_new_thread(servo_thread, ()) while True: time.sleep( 1 ) else: print( "> connected" ) status = wlan.ifconfig() print( "ip = " + status[0] ) # open socket addr = socket.getaddrinfo( "0.0.0.0", 80)[0][-1] s = socket.socket() s.setsockopt( socket.SOL_SOCKET, socket.SO_REUSEADDR, 1 ) #s.settimeout(1) s.bind( addr ) s.listen( 1 ) print( "> web server listening on", addr ) # listen for connections while True: print( ">" ) try: cl, addr = s.accept() print( "client connected from", addr) request = cl.recv( 1024 ) request = request.decode( "utf-8" ).strip() print( request ) if request.startswith( "GET / " ): print( "get data" ) response = html.replace( "<TEMPERATURE>", str(temperature) ) cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) cl.send( response ) elif request.startswith( "GET /min_temperature " ): print( "get min_temperature" ) cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) cl.send( str(temperature_min) ) elif request.startswith( "GET /max_temperature " ): print( "get max_temperature" ) cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) cl.send( str(temperature_max) ) elif request.startswith( "PUT /min_temperature " ): print( "put min_temperature" ) new_min_temperature = int(request.split( "\r\n\r\n" )[1]) print( new_min_temperature ) temperature_min = new_min_temperature cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) cl.send( "\"ok\"" ) elif request.startswith( "PUT /max_temperature " ): print( "put max_temperature" ) new_max_temperature = int(request.split( "\r\n\r\n" )[1]) print( new_max_temperature ) temperature_max = new_max_temperature cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) cl.send( "\"ok\"" ) elif request.startswith( "PUT /start " ): print( "put start" ) cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) if keep_going: print( " already started" ) cl.send( "\"already started\"" ) else: keep_going = True _thread.start_new_thread(servo_thread, ()) cl.send( "\"started\"" ) for i in range(5000, 6000): time.sleep( 0.001 ) servo_at = i servo.duty_u16( i ) elif request.startswith( "PUT /stop " ): print( "put stop" ) #cl.send( "HTTP/1.0 501 OK\r\nContent-type: application/json\r\n\r\n" ) #cl.send( "\"not implemented\"" ) # crash on stop, thread support is that bad cl.send( "HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n" ) if keep_going: keep_going = False cl.send( "\"stopped\"" ) else: print( " already stopped" ) cl.send( "\"already stopped\"" ) cl.close() except OSError as e: cl.close() print( "> connection closed" ) keep_going = False time.sleep( 2 ) except KeyboardInterrupt: print( "> ctrl+c, wrapping up..." ) keep_going = False time.sleep( 10 ) except Exception as e: print( e ) print( "> unexpected exception, wrapping up..." ) keep_going = False time.sleep( 10 )
max6675.pyimport time class MAX6675: MEASUREMENT_PERIOD_MS = 220 def __init__(self, sck, cs, so): """ Creates new object for controlling MAX6675 :param sck: SCK (clock) pin, must be configured as Pin.OUT :param cs: CS (select) pin, must be configured as Pin.OUT :param so: SO (data) pin, must be configured as Pin.IN """ # Thermocouple self._sck = sck self._sck.low() self._cs = cs self._cs.high() self._so = so self._so.low() self._last_measurement_start = 0 self._last_read_temp = 0 self._error = 0 def _cycle_sck(self): self._sck.high() time.sleep_us(1) self._sck.low() time.sleep_us(1) def refresh(self): """ Start a new measurement. """ self._cs.low() time.sleep_us(10) self._cs.high() self._last_measurement_start = time.ticks_ms() def ready(self): """ Signals if measurement is finished. :return: True if measurement is ready for reading. """ return time.ticks_ms() - self._last_measurement_start > MAX6675.MEASUREMENT_PERIOD_MS def error(self): """ Returns error bit of last reading. If this bit is set (=1), there's problem with the thermocouple - it can be damaged or loosely connected :return: Error bit value """ return self._error def read(self): """ Reads last measurement and starts a new one. If new measurement is not ready yet, returns last value. Note: The last measurement can be quite old (e.g. since last call to `read`). To refresh measurement, call `refresh` and wait for `ready` to become True before reading. :return: Measured temperature """ # Check if new reading is available if self.ready(): # Bring CS pin low to start protocol for reading result of # the conversion process. Forcing the pin down outputs # first (dummy) sign bit 15. self._cs.low() time.sleep_us(10) # Read temperature bits 14-3 from MAX6675. value = 0 for i in range(12): # SCK should resemble clock signal and new SO value # is presented at falling edge self._cycle_sck() value += self._so.value() << (11 - i) # Read the TC Input pin to check if the input is open self._cycle_sck() self._error = self._so.value() # Read the last two bits to complete protocol for i in range(2): self._cycle_sck() # Finish protocol and start new measurement self._cs.high() self._last_measurement_start = time.ticks_ms() self._last_read_temp = value * 0.25 return self._last_read_temp