test.py

#!/usr/bin/python3
# coding: utf8

import time

import RPi.GPIO as GPIO
from simple_pid import PID

import max6675
import ssr

GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)

sample_time = 4
setpoint = 28

cs1 = 4
cs2 = 5
sck = 24
so = 25

ssrA = 17
ssrB = 21

# power percentage: a+b should be 100%
percentageA = 50
percentageB = 50


def safe_off(pin):
    try:
        ssr.off(pin)
    except RuntimeError:
        pass


def emergency_stop(message):
    print("EMERGENCY STOP: {0}".format(message))
    safe_off(ssrA)
    safe_off(ssrB)


def validate_configuration():
    total_power = percentageA + percentageB
    if abs(total_power - 100) > 0.001:
        raise ValueError(
            "Invalid power split: percentageA + percentageB must be 100, got {0}".format(
                total_power
            )
        )
    if sample_time <= 0:
        raise ValueError("sample_time must be > 0")


def build_pid():
    pid_a = PID(5, 0.01, 0.1, setpoint=setpoint)
    pid_b = PID(5, 0.01, 0.1, setpoint=setpoint)

    for controller in (pid_a, pid_b):
        controller.output_limits = (0, 100)
        controller.sample_time = sample_time

    return pid_a, pid_b


def run_control_cycle(control_a, control_b):
    cycle_start = time.monotonic()
    cycle_end = cycle_start + sample_time

    channels = [
        {"pin": ssrA, "duty_cycle": control_a},
        {"pin": ssrB, "duty_cycle": control_b},
    ]

    for channel in channels:
        duty_cycle = max(0.0, min(channel["duty_cycle"], 100.0))
        on_duration = sample_time * duty_cycle / 100.0
        channel["off_time"] = cycle_start + on_duration
        channel["active"] = on_duration > 0

        if channel["active"]:
            ssr.on(channel["pin"])
        else:
            ssr.off(channel["pin"])

    while True:
        now = time.monotonic()
        if now >= cycle_end:
            break

        next_event = cycle_end - now
        for channel in channels:
            if channel["active"] and now >= channel["off_time"]:
                ssr.off(channel["pin"])
                channel["active"] = False

            if channel["active"]:
                next_event = min(next_event, channel["off_time"] - now)

        if next_event > 0:
            time.sleep(min(0.05, next_event))

    for channel in channels:
        if channel["active"]:
            ssr.off(channel["pin"])


def setup_hardware():
    max6675.set_pin(cs1, sck, so, max6675.UNIT_CELSIUS)
    max6675.set_pin(cs2, sck, so, max6675.UNIT_CELSIUS)
    ssr.set_pin(ssrA)
    ssr.set_pin(ssrB)


def main():
    validate_configuration()
    setup_hardware()
    pidA, pidB = build_pid()

    while True:
        tempA = max6675.read_temp(cs1)
        tempB = max6675.read_temp(cs2)

        controlA = pidA(tempA) * percentageA / 100.0
        controlB = pidB(tempB) * percentageB / 100.0

        print(
            "Thermocouple A Temperature: {0:0.1F}°C, control: {1:0.0F}%".format(
                tempA, controlA
            )
        )
        print(
            "Thermocouple B Temperature: {0:0.1F}°C, control: {1:0.0F}%".format(
                tempB, controlB
            )
        )

        run_control_cycle(controlA, controlB)


if __name__ == "__main__":
    exit_code = 0
    try:
        main()
    except KeyboardInterrupt:
        print("Controller stopped by user.")
    except max6675.Max6675Error as exc:
        exit_code = 1
        emergency_stop("sensor failure - {0}".format(exc))
    except Exception as exc:
        exit_code = 1
        emergency_stop("unexpected failure - {0}".format(exc))
    finally:
        safe_off(ssrA)
        safe_off(ssrB)
        GPIO.cleanup()

    raise SystemExit(exit_code)