kopia lustrzana https://github.com/jamesgao/kiln_controller
Add a bunch of simulation code to test PID
James Gao
rodzic
0d30eb02f4
commit
d1e295b85e
|
|
@ -1,47 +1,64 @@
|
|||
import stepper
|
||||
import datetime
|
||||
import numpy as np
|
||||
import time
|
||||
import random
|
||||
import warnings
|
||||
|
||||
class KilnController(object):
|
||||
""" schedule is of the format:
|
||||
[(time1, temp1), (time2, temp2), .... (timen, tempn)]
|
||||
where time is in seconds from start of firing and temp is in C"""
|
||||
|
||||
def __init__(self, schedule, monitor, interval=None):
|
||||
def __init__(self, schedule, monitor, interval=5, start_time=None, Kp=.01, Ki=.001, Kd=.001, simulate=True):
|
||||
self.schedule = schedule
|
||||
if interval is not None: # interpolate schedule at specified interval size
|
||||
temp = np.array(self.schedule)
|
||||
temp2 = []
|
||||
for i in range(len(self.schedule)-1):
|
||||
times = np.arange(self.schedule[i][0], self.schedule[i+1][0], interval)
|
||||
temps = np.arange(self.schedule[i][1], self.schedule[i+1][1], interval)
|
||||
temp2.extend([(times[j], temps[j]) for j in range(len(times))])
|
||||
self.schedule = temp2
|
||||
self.start_time = datetime.datetime.now()
|
||||
self.setpoint = schedule.pop(0)
|
||||
self.monitor = monitor
|
||||
self.pid = PID(3.0,0.4,1.2)
|
||||
self.regulator = stepper.Regulator()
|
||||
self.regulator.start()
|
||||
self.interval = interval
|
||||
self.start_time = start_time
|
||||
if start_time is None:
|
||||
self.start_time = time.time()
|
||||
self.regulator = stepper.Regulator(simulate=simulate)
|
||||
self.pid = PID(Kp, Ki, Kd)
|
||||
self.simulate = simulate
|
||||
if simulate:
|
||||
self.schedule.insert(0, [0, 15])
|
||||
else:
|
||||
self.schedule.insert(0, [0, self.monitor.temperature])
|
||||
|
||||
@property
|
||||
def elapsed(self):
|
||||
''' Returns the elapsed time from start in seconds'''
|
||||
return datetime.datetime.now() - self.start_time
|
||||
return time.time() - self.start_time
|
||||
|
||||
def run(self):
|
||||
self.pid.setPoint(self.setpoint[1])
|
||||
while self.elapsed<self.setpoint[0]:
|
||||
pid_out = self.pid.update(self.monitor.temperature)
|
||||
if pid_out<0 or pid_out>1:
|
||||
warnings.warn('PID output out of range at ' + str(pid_out) + '!')
|
||||
if pid_out < 0: pid_out = 0
|
||||
if pid_out > 1: pid_out = 1
|
||||
self.regulator.set(pid_out)
|
||||
self.setpoint = self.schedule.pop(0)
|
||||
if len(schedule)>0:
|
||||
self.run()
|
||||
try:
|
||||
self.regulator.ignite()
|
||||
print self.elapsed, self.schedule[-1][0]
|
||||
while self.elapsed < self.schedule[-1][0]:
|
||||
now = time.time()
|
||||
ts = self.elapsed
|
||||
#find epoch
|
||||
for i in range(len(self.schedule)-1):
|
||||
if self.schedule[i][0] < ts < self.schedule[i+1][0]:
|
||||
print "In epoch %d"%i
|
||||
time0, temp0 = self.schedule[i]
|
||||
time1, temp1 = self.schedule[i+1]
|
||||
frac = (ts - time0) / (time1 - time0)
|
||||
setpoint = frac * (temp1 - temp0) + temp0
|
||||
self.pid.setPoint(setpoint)
|
||||
|
||||
if self.simulate:
|
||||
temp = setpoint + random.gauss(-20, 15)
|
||||
else:
|
||||
temp = self.monitor.temperature
|
||||
|
||||
pid_out = self.pid.update(temp)
|
||||
if pid_out < 0: pid_out = 0
|
||||
if pid_out > 1: pid_out = 1
|
||||
self.regulator.set(pid_out)
|
||||
|
||||
time.sleep(self.interval - (time.time()-now))
|
||||
except:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print "Started at %f"%self.start_time
|
||||
|
||||
self.regulator.off()
|
||||
|
||||
class PID(object):
|
||||
"""
|
||||
|
|
@ -93,9 +110,10 @@ class PID(object):
|
|||
self.Integrator = self.Integrator_min
|
||||
|
||||
self.I_value = self.Integrator * self.Ki
|
||||
|
||||
PID = self.P_value + self.I_value + self.D_value
|
||||
|
||||
print "PID: %f, %f, %f: %f"%(self.P_value, self.I_value, self.D_value, PID)
|
||||
|
||||
return PID
|
||||
|
||||
def setPoint(self,set_point):
|
||||
|
|
|
|||
|
|
@ -4,7 +4,10 @@ import threading
|
|||
import warnings
|
||||
import Queue
|
||||
|
||||
from RPi import GPIO
|
||||
try:
|
||||
from RPi import GPIO
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
class Stepper(threading.Thread):
|
||||
pattern = [
|
||||
|
|
@ -49,26 +52,29 @@ class Stepper(threading.Thread):
|
|||
self.finished.wait()
|
||||
|
||||
def run(self):
|
||||
target = 0
|
||||
step, speed, block = self.queue.get()
|
||||
while step is not None:
|
||||
for pin, out in zip(self.pins, self.pattern[self.phase%len(self.pattern)]):
|
||||
GPIO.output(pin, out)
|
||||
try:
|
||||
step, speed, block = self.queue.get()
|
||||
while step is not None:
|
||||
for pin, out in zip(self.pins, self.pattern[self.phase%len(self.pattern)]):
|
||||
GPIO.output(pin, out)
|
||||
|
||||
if block:
|
||||
self._step(step, speed)
|
||||
self.finished.set()
|
||||
else:
|
||||
self.finished.set()
|
||||
self._step_noblock(step, speed)
|
||||
if block:
|
||||
self._step(step, speed)
|
||||
self.finished.set()
|
||||
else:
|
||||
self.finished.set()
|
||||
self._step_noblock(step, speed)
|
||||
|
||||
try:
|
||||
step, speed, block = self.queue.get(True, self.timeout)
|
||||
except Queue.Empty:
|
||||
#handle the timeout, turn off all pins
|
||||
for pin in self.pins:
|
||||
GPIO.output(pin, False)
|
||||
step, speed, block = self.queue.get()
|
||||
try:
|
||||
step, speed, block = self.queue.get(True, self.timeout)
|
||||
except Queue.Empty:
|
||||
#handle the timeout, turn off all pins
|
||||
for pin in self.pins:
|
||||
GPIO.output(pin, False)
|
||||
step, speed, block = self.queue.get()
|
||||
except:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
for pin in self.pins:
|
||||
GPIO.output(pin, False)
|
||||
|
|
@ -117,9 +123,21 @@ class Stepper(threading.Thread):
|
|||
|
||||
self.phase += step
|
||||
|
||||
class StepperSim(object):
|
||||
def __init__(self):
|
||||
self.phase = 0
|
||||
|
||||
def step(self, num, speed=10, block=False):
|
||||
print "Simulated stepping %d steps at %d steps / second"%(num, speed)
|
||||
if block:
|
||||
time.sleep(1)
|
||||
|
||||
def stop(self):
|
||||
print "stopping"
|
||||
|
||||
|
||||
class Regulator(object):
|
||||
def __init__(self, maxsteps=4500, minsteps=2500, speed=150, ignite_pin=26):
|
||||
def __init__(self, maxsteps=4500, minsteps=2400, speed=150, ignite_pin=None, simulate=False):
|
||||
"""Set up a stepper-controlled regulator. Implement some safety measures
|
||||
to make sure everything gets shut off at the end
|
||||
|
||||
|
|
@ -134,8 +152,11 @@ class Regulator(object):
|
|||
ignite_pin : int or None
|
||||
If not None, turn on this pin during the ignite sequence
|
||||
"""
|
||||
self.stepper = Stepper()
|
||||
self.stepper.start()
|
||||
if simulate:
|
||||
self.stepper = StepperSim()
|
||||
else:
|
||||
self.stepper = Stepper()
|
||||
self.stepper.start()
|
||||
self.current = 0
|
||||
self.max = maxsteps
|
||||
self.min = minsteps
|
||||
|
|
@ -146,11 +167,13 @@ class Regulator(object):
|
|||
GPIO.setup(ignite_pin, OUT)
|
||||
|
||||
def exit():
|
||||
self.off()
|
||||
if self.current != 0:
|
||||
self.off()
|
||||
self.stepper.stop()
|
||||
atexit.register(exit)
|
||||
|
||||
def ignite(self, start=2800, delay=5):
|
||||
def ignite(self, start=2800, delay=1):
|
||||
print "Ignition..."
|
||||
self.stepper.step(start, self.speed, block=True)
|
||||
if self.ignite_pin is not None:
|
||||
GPIO.output(self.ignite_pin, True)
|
||||
|
|
@ -159,12 +182,17 @@ class Regulator(object):
|
|||
GPIO.output(self.ignite_pin, False)
|
||||
self.stepper.step(self.min - start, self.speed)
|
||||
self.current = self.min
|
||||
print "Done!"
|
||||
|
||||
def off(self, block=True):
|
||||
print "Turning off..."
|
||||
self.stepper.step(-self.current, self.speed, block=block)
|
||||
self.current = 0
|
||||
print "Done!"
|
||||
|
||||
def set(self, value, block=False):
|
||||
if self.current == 0:
|
||||
raise ValueError("System must be ignited to set value")
|
||||
if not 0 <= value <= 1:
|
||||
raise ValueError("Must give fraction between 0 and 1")
|
||||
target = int(value * (self.max - self.min) + self.min)
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue