<s> from twisted . internet . defer import inlineCallbacks , returnValue <EOL> from twisted . python . constants import ValueConstant , Values <EOL> from twisted . python . reflect import namedObject <EOL> import smartanthill . network . cdc as cdc <EOL> from smartanthill . device . arg import DeviceIDArg <EOL> from smartanthill . network . zvd import ZeroVirtualDevice <EOL> class OperationType ( Values ) : <EOL> PING = ValueConstant ( <EOL> cdc . CHANNEL_URGENT . PING . value ) <EOL> LIST_OPERATIONS = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . LIST_OPERATIONS . value ) <EOL> CONFIGURE_PIN_MODE = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . CONFIGURE_PIN_MODE . value ) <EOL> READ_DIGITAL_PIN = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . READ_DIGITAL_PIN . value ) <EOL> WRITE_DIGITAL_PIN = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . WRITE_DIGITAL_PIN . value ) <EOL> CONFIGURE_ANALOG_REFERENCE = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . CONFIGURE_ANALOG_REFERENCE . value ) <EOL> READ_ANALOG_PIN = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . READ_ANALOG_PIN . value ) <EOL> WRITE_ANALOG_PIN = ValueConstant ( <EOL> cdc . CHANNEL_BDCREQUEST . WRITE_ANALOG_PIN . value ) <EOL> class OperationBase ( object ) : <EOL> TYPE = None <EOL> TTL = <NUM_LIT:1> <EOL> ACK = True <EOL> REQUIRED_PARAMS = None <EOL> def __init__ ( self , board , data ) : <EOL> self . board = board <EOL> self . data = data <EOL> def process_data ( self , data ) : <EOL> return [ ] <EOL> def on_result ( self , result ) : <EOL> return result <EOL> @ inlineCallbacks <EOL> def launch ( self , devid ) : <EOL> devidarg = DeviceIDArg ( ) <EOL> devidarg . set_value ( devid ) <EOL> zvd = ZeroVirtualDevice ( ) <EOL> result = yield zvd . request ( self . TYPE . value , devidarg . get_value ( ) , <EOL> self . TTL , self . ACK , <EOL> self . process_data ( self . data ) ) <EOL> returnValue ( self . on_result ( result ) ) <EOL> def check_params ( self , params ) : <EOL> if not self . REQUIRED_PARAMS : <EOL> return True <EOL> params = [ s if "<STR_LIT:[>" not in s else s [ : s . find ( "<STR_LIT:[>" ) ] + "<STR_LIT>" for s in params ] <EOL> return set ( self . REQUIRED_PARAMS ) <= set ( params ) <EOL> def get_operation_class ( type_ ) : <EOL> assert isinstance ( type_ , ValueConstant ) <EOL> if not hasattr ( get_operation_class , "<STR_LIT>" ) : <EOL> get_operation_class . cache = { } <EOL> if type_ . value not in get_operation_class . cache : <EOL> get_operation_class . cache [ type_ . value ] = namedObject ( <EOL> "<STR_LIT>" % type_ . name . lower ( ) ) <EOL> return get_operation_class . cache [ type_ . value ] </s>
