Control sequence for multiple hydrogenation experiments

DoE.py

from octopus.transport.basic import tcp
from octopus.manufacturer import vapourtec, thalesnano, mt, knauer, vici

from octopus.runtime import *

h = thalesnano.HCube(tcp("192.168.15.156", 9004))
v = vici.MultiValve(tcp("192.168.15.153", 9006))
k = knauer.K120(tcp("192.168.15.156", 9002))
ir = mt.ICIR(tcp("192.168.15.242", 8124), stream_names = ["starting_material", "product"])

id("hydrogenation_opt_1")
title("Hydrogenation")

setup = sequence(
	log("Preparing..."),
	set(v.position, 1),
	set(k.target, 200),
	set(k.power, "on"),
	set(h.column_temperature_target, 25),
	set(h.system_pressure_target, 0),
	set(h.hydrogen_mode, "full"),
	call(h.start_hydrogenation)
)

dead_volume = 3000 # uL
current_state = [25, "full", 200] # temp, pressure, rate

def reaction (name, pressure, temp, rate, valve_pos):
	global current_state 

	old_state, current_state = current_state, [temp, pressure, rate]

	change_t = (old_state[0] != temp)
	change_p = (old_state[1] != pressure)
	change_r = (old_state[2] != rate)

	if change_p:
		if pressure == "full":
			change_h2 = sequence(
				set(h.system_pressure_target, 0),
				set(h.hydrogen_mode, "full"),
			)

			press_wait = (h.state == "running")
		else:
			change_h2 = sequence(
				set(h.hydrogen_mode, "controlled"),
				set(h.system_pressure_target, pressure),
			)

			press_wait = (h.system_pressure >= pressure)

		change_step = sequence(
			log("Switch off hydrogen to change pressure"),
			call(h.stop_keep_hydrogen),
			log("Change pressure to %s ..." % pressure),
			change_h2,
			log("Change temp to %s" % temp),
			set(h.column_temperature_target, temp),
			log("Change rate to %s" % rate),
			set(k.target, rate),
			log("Starting H-Cube"),
			parallel(
				wait(((dead_volume / rate) + 1) * 60),
				sequence(
					call(h.start_hydrogenation),
					do_if(h.state != "running",
						log("Please check that the H-cube is running OK")
					),
					wait_until(press_wait),
				),
			),
		)
	else:
		if change_r:
			a = sequence(
				log("Change rate to %s" % rate),
				set(k.target, rate),
				log("Wait one dead volume"),
				wait(((dead_volume / rate) + 1) * 60),
			)
		else:
			a = log("Rate not changed")

		if change_t:
			if temp > old_state[0]:
				# Wait to heat up
				temp_wait = h.column_temperature >= temp
			else:
				# Wait to cool down
				temp_wait = h.column_temperature < temp + 5

			b = sequence(
				log("Change temp to %s" % temp),
				set(h.column_temperature_target, temp),
				log("Waiting for temp. stability"),
				wait_until(temp_wait),
			)
		else:
			b = log("Temp not changed")

		change_step = parallel(a, b)

	return sequence(
		log_output(name),
		log("Preparing reaction #%s" % name),
		change_step,
		log("Starting reaction #%s" % name),
		wait(((dead_volume / rate) + 1) * 60),
		log("Taking a sample now"),
		set(v.position, valve_pos),
		wait("10m"),
		log("Collection complete"),
		set(v.position, 1),
	)

ui(
	traces =  [{
		"title": "Pressures",
		"unit": "",
		"traces": [h.inlet_pressure, h.system_pressure],
		"maxtime": 30 * 60
	}, {
		"title": "IR output",
		"unit": "mAU",
		"traces": [ir.starting_material, ir.product],
		"maxtime": 30 * 60
	}],
	properties = [
		k.rate, 
		h.hydrogen_mode, 
		h.system_pressure_target, 
		h.system_pressure, 
		h.column_temperature_target, 
		h.column_temperature,
		h.state,
		h.message,
		v.position
	]
)

run(sequence(
	setup,

	## First Run:

	reaction(1 , "full",  100,	100, 2),
	reaction(2 , 20	   ,  100,	100, 3),
	reaction(3 , 20	   ,  40, 	200, 4),
	reaction(4 , "full",  40, 	100, 5),
	reaction(5 , 20	   ,  40, 	200, 6),
	reaction(6 , 20	   ,  40, 	100, 7),
	reaction(7 , 20	   ,  100,	200, 8),
	reaction(8 , "full",  40, 	200, 9),

	## Second Run:

	#reaction(9 , 20	,  40, 	100, 2),
	#reaction(10, "full",  100,	100, 3),
	#reaction(11, "full",  40, 	100, 4),
	#reaction(12, "full",  100,	200, 5),
	#reaction(13, "full",  100,	200, 6),
	#reaction(14, 20	,  100,	100, 7),
	#reaction(15, "full",  40, 	200, 8),
	#reaction(16, 20    ,  100,	200, 9),

	log("Complete, releasing hydrogen"),
	call(h.stop_release_hydrogen),

	log("Flushing"),
	set(k.target, 200),
	wait("20m"),
	set(k.power, "off")
))