guest_worker/sorteerhoed/statemachine.py

89 lines
3 KiB
Python

import datetime
class State():
def __init__(self, hit_id):
self.time = datetime.datetime.now()
self.hit_id = params['hit_id']
def transition(self, transitionName, params = {}):
raise Exception("Not implemented")
class StateMachine:
def __init__(self, initalState):
self.history = [('init',initialState)]
def current(self):
return self.history[-1][1]
def transition(self, transitionName, params):
# TODO: update Store & Interface
if transitionName not in self.current().availableTransitions:
raise Exception("Invalid transition")
newState = self.current().transition(transitionName, params)
if not newState:
raise RuntimeException(f"Invalid transition {transitionName} for {self.current()}")
self.history.append((transitionName, newState))
def getStateForHit(self, hit_id, stateCls = None):
states = [s for s in self.history if s[1].hit_id == hit_id and (stateCls is None or isinstance(s[1], stateCls))]
if len(states < 1):
return None
return states[-1]
class HITCreated(State):
availableTransitions = ['accept']
self.state = None
self.fee = None
self.hit_created = None
self.hit_opened = None
self.hit_submitted = None
def transition(self, transitionName, params = {}):
if transitionName == 'accept':
return HITAssigned(params['hit_id'], params['assignment_id'])
class HITAssigned(State):
availableTransitions = ['reject', 'abandon', 'submit']
def __init__(self, hit_id):
self.assignment_id = None
self.worker_id = None
self.ip = None
self.location = None
self.browser = None
self.os = None
self.resolution = None
def transition(self, transitionName, params = {}):
if transitionName == 'reject' or transitionName == 'abandon':
return HITAbandonedRejected(params['hit_id'])
if transitionName == 'submit':
return HITSubmitted(params['hit_id'])
class HITAbandonedRejected(State):
availableTransitions = ['accept']
def transition(self, transitionName, params = {}):
if transitionName == 'accept':
return HITAssigned(params['hit_id'])
class HITSubmitted(State):
availableTransitions = ['scan']
def transition(self, transitionName, params = {}):
if transitionName == 'scan':
return Scanning(params['hit_id'])
class Scanning(State):
availableTransitions = ['scan_complete', 'scan_failed']
def transition(self, transitionName, params = {}):
if transitionName == 'scan_complete':
return ImageAvailable(params['hit_id'])
if transitionName == 'scan_failed':
raise Exception("Scan failed, unknown state")
class ImageAvailable(State):
availableTransitions = ['create_hit']
def transition(self, transitionName, params = {}):
if transitionName == 'create_hit':
return HITCreated(params['hit_id'])