Error Recovery: Script Example

PyHamilton
1

Thought I’d share a really simple example of error recovery so folks could see how to implement it in their scripts. Let me know if this is something you find useful!

This is basically just the template script that you create with pyhamilton-new-project with some small functions added

active_column = 0
recovery_state = {'active_column': active_column}

def recover_state(file):
    with open(file, "r") as read_content:
        state = json.load(read_content)
    return state

def write_state(file, state):
    with open(file, "w") as write_file:
        json.dump(state, write_file)

if '--error-recovery' in sys.argv:
    recovered_state = recover_state('state.json')
    active_column = recovered_state['active_column']

and inside the command loop we wish to recover from:

active_column += 1
recovery_state.update({'active_column': active_column})
write_state('state.json', recovery_state)

The whole script together:

import os
from pyhamilton import (HamiltonInterface,  LayoutManager, 
 Plate96, Tip96, initialize, tip_pick_up, tip_eject, 
 aspirate, dispense,  oemerr, resource_list_with_prefix, normal_logging)
import json
import sys

liq_class = 'StandardVolumeFilter_Water_DispenseJet_Empty'

lmgr = LayoutManager('deck.lay')
plates = resource_list_with_prefix(lmgr, 'plate_', Plate96, 5)
tips = resource_list_with_prefix(lmgr, 'tips_', Tip96, 1)
liq_class = 'StandardVolumeFilter_Water_DispenseJet_Empty'

number_columns = 12
aspiration_poss_cols = [[(plates[1], well) for well in range(8*column, 8*column+8)] for column in range(number_columns)]
dispense_poss_cols = [[(plates[0], well) for well in range(8*column, 8*column+8)] for column in range(number_columns)]
vols_list = [30]*8


tips_poss_cols = [[(tips[0], well) for well in range(8*column, 8*column+8)] for column in range(12)]

active_column = 0
recovery_state = {'active_column': active_column}

def recover_state(file):
    with open(file, "r") as read_content:
        state = json.load(read_content)
    return state

def write_state(file, state):
    with open(file, "w") as write_file:
        json.dump(state, write_file)

if '--error-recovery' in sys.argv:
    recovered_state = recover_state('state.json')
    active_column = recovered_state['active_column']

if __name__ == '__main__': 
    with HamiltonInterface(simulate=True) as ham_int:
        normal_logging(ham_int, os.getcwd())
        initialize(ham_int)
        while active_column <= 12:
            tip_pick_up(ham_int, tips_poss_cols[active_column])
            aspirate(ham_int, aspiration_poss_cols[active_column], vols_list, liquidClass = liq_class)
            dispense(ham_int, dispense_poss_cols[active_column], vols_list, liquidClass = liq_class)
            tip_eject(ham_int, tips_poss_cols[active_column])
            
            active_column += 1
            recovery_state.update({'active_column': active_column})
            write_state('state.json', recovery_state)

GIF of Script Simulation
error_recovery

6 Likes
2

Hey Stefan. 1st off, want to say how much I admire/love your work on this, I don’t have the right words but it’s incredible!!

I’m in the midst of working on a project to use your PyHamilton here at our lab (otherwise we just use the regular Venus software clearly) for both Microlab STARS & Vantages. I got the example scripts to open run control (although they don’t run the method automatically, I’ll clearly have to add in more for it to do that) BUT I’m definitely trying to incorporate the error recovery as well. As you know the software doesn’t provide any method recovery once you encounter unrecoverable errors & have to abort runs so it’d be a HUGE life-saver to make method recovery possible through this so I’m hoping by figuring out your error recovery script & building off that to make that possible.

Since you’re clearly the master at this I have to ask in hopes your brain can spread some wisdom to help… do you happen to know of any example scripts or how it’d be possible to make method recovery a reality through PyHamilton? At least just for one method in general like the example method you provided? The dream would be to make a generic or universal program through this that can recover any existing methods with the same concepts of direct connect servers in addition to tracing processed steps & all that. Let me know when you can, I know it’s a lot but any advice would make the biggest difference.

Either way, this is incredible & I applaud you to say the least, thank you for this!

2 Likes
3

Hi,

Thanks for the very kind words, it is always great to hear.

It sounds like you want a universal error recovery system that accounts for every possible break point and state change rather than manually accounting for only a certain number of state changes.

This is definitely possible but will require some thoughtful design. You will basically want to index each individual step in a data structure and write the current index to a file so that you can recover from that file. You will also want to track any other variables that you would like to restart from, e.g. tips, plates, wells, samples, etc.

You will probably want to create this as a library that can be reused across new methods rather than reimplementing for every single method. If you want, we can have a call to discuss the problem in more detail.

I also recommend looking into decorators (I love decorators) as ways to wrap additional behavior around pre-existing functions. So you can have some base function e.g. aspirate that gets decorated with state tracking behavior.

2 Likes
4

… I can’t thank you enough for the advice. At risk of pushing my luck, I’m afraid I have more to ask.

I figured out how to do a lot with what you provided (including error recovery). However, applying it to preexisting methods without having to rewrite them through python is where I get lost. I can replicate your examples with .lay files from preexisting methods. We got it to produce a JSON for ‘active_column’ example but the example obviously shows how to send asp/disp commands which the ‘active column’ tracking covers most of what that example method did but for existing methods where there’s a lot more variable & sequences, I’m lost on how to apply the example of what existing methods do & how to retrieve that info in order to do anything within the py scripts. Is it possible to retrieve liquid classes, variables & sequences from an existing method through python? I’m trying to use the same foundation while incorporating these tools but also without having to change the existing methods too much. I was hoping through the trace files or the created recovery JSON file to be able to recover from there but I’m stuck on how to replicate the existing method to create a starting point to recover from (where sequences/steps/variables have been processed/used for the method) & how to go from there with that info with the recovery function (I get you’d establish variables in the py scripts like in the example). I’m hoping I make sense as it’s clearly a hard project in itself so it’s also hard to explain. Any help you can offer would be beyond appreciated!!!

1 Like
5

I forgot to mention, I also extract method name, sequences & variables from the existing methods trace & hsl files but still trying to figure out how to incorporate that as I said before in hopes that helps anything.

6

I’ll need more information in order to give more concrete advice relevant to your problem, I’ll message you with more questions.

Yes this is likely possible but probably not advisable or necessary in a fully automated way without really knowing what we are looking for ahead of time. Maybe we can try manually sketching out what we need for a first pass.

7

I’ll give a general solution:

You can take any script and split it up into functions that each take a set of tracked variables {a,b,c…} as inputs. Each function should return the same set of variables that the next function takes as inputs. So,

y = f(x)
g(y)

For simplicity you should have one variable like a dictionary or class that contains all of the tracked variables, so in the above example you would have f(x) update parameters inside x and then pass that to g(x). Before you return from f(x), you should write x to a JSON file.

Suppose you crash inside of g(x). Instead of starting again from f(x), you can use the variables you wrote to JSON to enter g(x). Now you just need some way to track what function to recover into. Add an additional variable to your x = {a,b,c,...} that says which function to enter. Make sure to iterate this counter at the end of every function.

Finally, encapsulate all of this into a function that loops over a list whose elements are the functions you broke your script up into. That will look something like this:

functions_list = [f,g,h]

def loop_over_functions(functions_list, state):
     for fun in range(starting_index, len(functions_list)):
        state = functions_list[state['index']](state)
        index += 1
        

if error_recovery:
     state = read_state(...)
else:
    state = initialize_variables(...)

loop_over_functions(...)

Note that in this line:
state = functions_list[state['index']](state)

[state[‘index’]] points to the function we are entering in the list. This assumes we are updating state[‘index’] inside the functions right before they return.