<s> from __future__ import division <EOL> import os <EOL> import sys <EOL> import argparse <EOL> import shutil <EOL> import socket <EOL> import time <EOL> from datetime import datetime , timedelta <EOL> from zope . interface import implements <EOL> from twisted . protocols . basic import LineReceiver <EOL> from twisted . internet . protocol import Factory , Protocol <EOL> from twisted . internet import reactor , interfaces <EOL> from twisted . internet . error import ConnectionLost , ConnectionDone <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT:..>' ) ) <EOL> from daqpower import log <EOL> from daqpower . config import DeviceConfiguration <EOL> from daqpower . common import DaqServerRequest , DaqServerResponse , Status <EOL> try : <EOL> from daqpower . daq import DaqRunner , list_available_devices , CAN_ENUMERATE_DEVICES <EOL> __import_error = None <EOL> except ImportError as e : <EOL> __import_error = e <EOL> DaqRunner = None <EOL> list_available_devices = lambda : [ '<STR_LIT>' ] <EOL> class ProtocolError ( Exception ) : <EOL> pass <EOL> class DummyDaqRunner ( object ) : <EOL> """<STR_LIT>""" <EOL> num_rows = <NUM_LIT:200> <EOL> @ property <EOL> def number_of_ports ( self ) : <EOL> return self . config . number_of_ports <EOL> def __init__ ( self , config , output_directory ) : <EOL> log . info ( '<STR_LIT>' . format ( config , output_directory ) ) <EOL> self . config = config <EOL> self . output_directory = output_directory <EOL> self . is_running = False <EOL> def start ( self ) : <EOL> import csv , random <EOL> log . info ( '<STR_LIT>' ) <EOL> for i in xrange ( self . config . number_of_ports ) : <EOL> rows = [ [ '<STR_LIT>' , '<STR_LIT>' ] ] + [ [ random . gauss ( <NUM_LIT:1.0> , <NUM_LIT:1.0> ) , random . gauss ( <NUM_LIT:1.0> , <NUM_LIT:0.1> ) ] <EOL> for _ in xrange ( self . num_rows ) ] <EOL> with open ( self . get_port_file_path ( self . config . labels [ i ] ) , '<STR_LIT:wb>' ) as wfh : <EOL> writer = csv . writer ( wfh ) <EOL> writer . writerows ( rows ) <EOL> self . is_running = True <EOL> def stop ( self ) : <EOL> self . is_running = False <EOL> log . info ( '<STR_LIT>' ) <EOL> def get_port_file_path ( self , port_id ) : <EOL> if port_id in self . config . labels : <EOL> return os . path . join ( self . output_directory , '<STR_LIT>' . format ( port_id ) ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' . format ( port_id ) ) <EOL> class DaqServer ( object ) : <EOL> def __init__ ( self , base_output_directory ) : <EOL> self . base_output_directory = os . path . abspath ( base_output_directory ) <EOL> if os . path . isdir ( self . base_output_directory ) : <EOL> log . info ( '<STR_LIT>' . format ( self . base_output_directory ) ) <EOL> else : <EOL> log . info ( '<STR_LIT>' . format ( self . base_output_directory ) ) <EOL> os . makedirs ( self . base_output_directory ) <EOL> self . runner = None <EOL> self . output_directory = None <EOL> self . labels = None <EOL> def configure ( self , config_string ) : <EOL> message = None <EOL> if self . runner : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> if self . runner . is_running : <EOL> self . runner . stop ( ) <EOL> config = DeviceConfiguration . deserialize ( config_string ) <EOL> config . validate ( ) <EOL> self . output_directory = self . _create_output_directory ( ) <EOL> self . labels = config . labels <EOL> log . info ( '<STR_LIT>' . format ( self . output_directory ) ) <EOL> self . runner = DaqRunner ( config , self . output_directory ) <EOL> return message <EOL> def start ( self ) : <EOL> if self . runner : <EOL> if not self . runner . is_running : <EOL> self . runner . start ( ) <EOL> else : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> self . runner . stop ( ) <EOL> self . runner . start ( ) <EOL> return message <EOL> else : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> def stop ( self ) : <EOL> if self . runner : <EOL> if self . runner . is_running : <EOL> self . runner . stop ( ) <EOL> else : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> self . runner . stop ( ) <EOL> return message <EOL> else : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> def list_devices ( self ) : <EOL> return list_available_devices ( ) <EOL> def list_ports ( self ) : <EOL> return self . labels <EOL> def list_port_files ( self ) : <EOL> if not self . runner : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> ports_with_files = [ ] <EOL> for port_id in self . labels : <EOL> path = self . get_port_file_path ( port_id ) <EOL> if os . path . isfile ( path ) : <EOL> ports_with_files . append ( port_id ) <EOL> return ports_with_files <EOL> def get_port_file_path ( self , port_id ) : <EOL> if not self . runner : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> return self . runner . get_port_file_path ( port_id ) <EOL> def terminate ( self ) : <EOL> message = None <EOL> if self . runner : <EOL> if self . runner . is_running : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> self . runner . stop ( ) <EOL> self . runner = None <EOL> if self . output_directory and os . path . isdir ( self . output_directory ) : <EOL> shutil . rmtree ( self . output_directory ) <EOL> self . output_directory = None <EOL> log . info ( '<STR_LIT>' ) <EOL> else : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> return message <EOL> def _create_output_directory ( self ) : <EOL> basename = datetime . now ( ) . strftime ( '<STR_LIT>' ) <EOL> dirname = os . path . join ( self . base_output_directory , basename ) <EOL> os . makedirs ( dirname ) <EOL> return dirname <EOL> def __del__ ( self ) : <EOL> if self . runner : <EOL> self . runner . stop ( ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . base_output_directory ) <EOL> __repr__ = __str__ <EOL> class DaqControlProtocol ( LineReceiver ) : <EOL> def __init__ ( self , daq_server ) : <EOL> self . daq_server = daq_server <EOL> self . factory = None <EOL> def lineReceived ( self , line ) : <EOL> line = line . strip ( ) <EOL> log . info ( '<STR_LIT>' . format ( line ) ) <EOL> try : <EOL> request = DaqServerRequest . deserialize ( line ) <EOL> except Exception , e : <EOL> message = getattr ( e , '<STR_LIT>' , e . message ) <EOL> self . sendError ( '<STR_LIT>' . format ( e . __class__ . __name__ , message ) ) <EOL> else : <EOL> self . processRequest ( request ) <EOL> def processRequest ( self , request ) : <EOL> try : <EOL> if request . command == '<STR_LIT>' : <EOL> self . configure ( request ) <EOL> elif request . command == '<STR_LIT:start>' : <EOL> self . start ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . stop ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . list_devices ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . list_ports ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . list_port_files ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . pull_port_data ( request ) <EOL> elif request . command == '<STR_LIT>' : <EOL> self . terminate ( request ) <EOL> else : <EOL> self . sendError ( '<STR_LIT>' . format ( request . command ) ) <EOL> except Exception , e : <EOL> message = getattr ( e , '<STR_LIT>' , e . message ) <EOL> self . sendError ( '<STR_LIT>' . format ( e . __class__ . __name__ , message ) ) <EOL> def configure ( self , request ) : <EOL> if '<STR_LIT>' in request . params : <EOL> result = self . daq_server . configure ( request . params [ '<STR_LIT>' ] ) <EOL> if not result : <EOL> self . sendResponse ( Status . OK ) <EOL> else : <EOL> self . sendResponse ( Status . OKISH , message = result ) <EOL> else : <EOL> self . sendError ( '<STR_LIT>' ) <EOL> def start ( self , request ) : <EOL> result = self . daq_server . start ( ) <EOL> if not result : <EOL> self . sendResponse ( Status . OK ) <EOL> else : <EOL> self . sendResponse ( Status . OKISH , message = result ) <EOL> def stop ( self , request ) : <EOL> result = self . daq_server . stop ( ) <EOL> if not result : <EOL> self . sendResponse ( Status . OK ) <EOL> else : <EOL> self . sendResponse ( Status . OKISH , message = result ) <EOL> def pull_port_data ( self , request ) : <EOL> if '<STR_LIT>' in request . params : <EOL> port_id = request . params [ '<STR_LIT>' ] <EOL> port_file = self . daq_server . get_port_file_path ( port_id ) <EOL> if os . path . isfile ( port_file ) : <EOL> port = self . _initiate_file_transfer ( port_file ) <EOL> self . sendResponse ( Status . OK , data = { '<STR_LIT>' : port } ) <EOL> else : <EOL> self . sendError ( '<STR_LIT>' . format ( port_id ) ) <EOL> else : <EOL> self . sendError ( '<STR_LIT>' ) <EOL> def list_devices ( self , request ) : <EOL> if CAN_ENUMERATE_DEVICES : <EOL> devices = self . daq_server . list_devices ( ) <EOL> self . sendResponse ( Status . OK , data = { '<STR_LIT>' : devices } ) <EOL> else : <EOL> message = "<STR_LIT>" <EOL> self . sendResponse ( Status . OKISH , message = message ) <EOL> def list_ports ( self , request ) : <EOL> port_labels = self . daq_server . list_ports ( ) <EOL> self . sendResponse ( Status . OK , data = { '<STR_LIT>' : port_labels } ) <EOL> def list_port_files ( self , request ) : <EOL> port_labels = self . daq_server . list_port_files ( ) <EOL> self . sendResponse ( Status . OK , data = { '<STR_LIT>' : port_labels } ) <EOL> def terminate ( self , request ) : <EOL> status = Status . OK <EOL> message = '<STR_LIT>' <EOL> if self . factory . transfer_sessions : <EOL> message = '<STR_LIT>' <EOL> log . warning ( message ) <EOL> for session in self . factory . transfer_sessions : <EOL> self . factory . transferComplete ( session ) <EOL> message += self . daq_server . terminate ( ) or '<STR_LIT>' <EOL> if message : <EOL> status = Status . OKISH <EOL> self . sendResponse ( status , message ) <EOL> def sendError ( self , message ) : <EOL> log . error ( message ) <EOL> self . sendResponse ( Status . ERROR , message ) <EOL> def sendResponse ( self , status , message = None , data = None ) : <EOL> response = DaqServerResponse ( status , message = message , data = data ) <EOL> self . sendLine ( response . serialize ( ) ) <EOL> def sendLine ( self , line ) : <EOL> log . info ( '<STR_LIT>' . format ( line ) ) <EOL> LineReceiver . sendLine ( self , line . replace ( '<STR_LIT:\r\n>' , '<STR_LIT>' ) ) <EOL> def _initiate_file_transfer ( self , filepath ) : <EOL> sender_factory = FileSenderFactory ( filepath , self . factory ) <EOL> connector = reactor . listenTCP ( <NUM_LIT:0> , sender_factory ) <EOL> self . factory . transferInitiated ( sender_factory , connector ) <EOL> return connector . getHost ( ) . port <EOL> class DaqFactory ( Factory ) : <EOL> protocol = DaqControlProtocol <EOL> check_alive_period = <NUM_LIT:5> * <NUM_LIT> <EOL> max_transfer_lifetime = <NUM_LIT:30> * <NUM_LIT> <EOL> def __init__ ( self , server , cleanup_period = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , cleanup_after_days = <NUM_LIT:5> ) : <EOL> self . server = server <EOL> self . cleanup_period = cleanup_period <EOL> self . cleanup_threshold = timedelta ( cleanup_after_days ) <EOL> self . transfer_sessions = { } <EOL> def buildProtocol ( self , addr ) : <EOL> proto = DaqControlProtocol ( self . server ) <EOL> proto . factory = self <EOL> reactor . callLater ( self . check_alive_period , self . pulse ) <EOL> reactor . callLater ( self . cleanup_period , self . perform_cleanup ) <EOL> return proto <EOL> def clientConnectionLost ( self , connector , reason ) : <EOL> log . msg ( '<STR_LIT>' . format ( reason ) ) <EOL> if not isinstance ( reason , ConnectionLost ) : <EOL> log . msg ( '<STR_LIT>' ) <EOL> for session in self . transfer_sessions : <EOL> self . transferComplete ( session ) <EOL> def transferInitiated ( self , session , connector ) : <EOL> self . transfer_sessions [ session ] = ( time . time ( ) , connector ) <EOL> def transferComplete ( self , session , reason = '<STR_LIT:OK>' ) : <EOL> if reason != '<STR_LIT:OK>' : <EOL> log . error ( reason ) <EOL> self . transfer_sessions [ session ] [ <NUM_LIT:1> ] . stopListening ( ) <EOL> del self . transfer_sessions [ session ] <EOL> def pulse ( self ) : <EOL> """<STR_LIT>""" <EOL> current_time = time . time ( ) <EOL> for session in self . transfer_sessions : <EOL> start_time , conn = self . transfer_sessions [ session ] <EOL> if ( current_time - start_time ) > self . max_transfer_lifetime : <EOL> message = '<STR_LIT>' <EOL> self . transferComplete ( session , message . format ( session , conn . getHost ( ) . port ) ) <EOL> if self . transfer_sessions : <EOL> reactor . callLater ( self . check_alive_period , self . pulse ) <EOL> def perform_cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> log . msg ( '<STR_LIT>' ) <EOL> base_directory = self . server . base_output_directory <EOL> current_time = datetime . now ( ) <EOL> for entry in os . listdir ( base_directory ) : <EOL> entry_path = os . path . join ( base_directory , entry ) <EOL> entry_ctime = datetime . fromtimestamp ( os . path . getctime ( entry_path ) ) <EOL> existence_time = current_time - entry_ctime <EOL> if existence_time > self . cleanup_threshold : <EOL> log . debug ( '<STR_LIT>' . format ( entry , existence_time ) ) <EOL> shutil . rmtree ( entry_path ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' . format ( entry , existence_time ) ) <EOL> log . msg ( '<STR_LIT>' ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . server ) <EOL> __repr__ = __str__ <EOL> class FileReader ( object ) : <EOL> implements ( interfaces . IPushProducer ) <EOL> def __init__ ( self , filepath ) : <EOL> self . fh = open ( filepath ) <EOL> self . proto = None <EOL> self . done = False <EOL> self . _paused = True <EOL> def setProtocol ( self , proto ) : <EOL> self . proto = proto <EOL> def resumeProducing ( self ) : <EOL> if not self . proto : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> self . _paused = False <EOL> try : <EOL> while not self . _paused : <EOL> line = self . fh . next ( ) . rstrip ( '<STR_LIT:\n>' ) + '<STR_LIT:\r\n>' <EOL> self . proto . transport . write ( line ) <EOL> except StopIteration : <EOL> log . debug ( '<STR_LIT>' ) <EOL> self . stopProducing ( ) <EOL> def pauseProducing ( self ) : <EOL> self . _paused = True <EOL> def stopProducing ( self ) : <EOL> self . done = True <EOL> self . fh . close ( ) <EOL> self . proto . transport . unregisterProducer ( ) <EOL> self . proto . transport . loseConnection ( ) <EOL> class FileSenderProtocol ( Protocol ) : <EOL> def __init__ ( self , reader ) : <EOL> self . reader = reader <EOL> self . factory = None <EOL> def connectionMade ( self ) : <EOL> self . transport . registerProducer ( self . reader , True ) <EOL> self . reader . resumeProducing ( ) <EOL> def connectionLost ( self , reason = ConnectionDone ) : <EOL> if self . reader . done : <EOL> self . factory . transferComplete ( ) <EOL> else : <EOL> self . reader . pauseProducing ( ) <EOL> self . transport . unregisterProducer ( ) <EOL> class FileSenderFactory ( Factory ) : <EOL> @ property <EOL> def done ( self ) : <EOL> if self . reader : <EOL> return self . reader . done <EOL> else : <EOL> return None <EOL> def __init__ ( self , path , owner ) : <EOL> self . path = os . path . abspath ( path ) <EOL> self . reader = None <EOL> self . owner = owner <EOL> def buildProtocol ( self , addr ) : <EOL> if not self . reader : <EOL> self . reader = FileReader ( self . path ) <EOL> proto = FileSenderProtocol ( self . reader ) <EOL> proto . factory = self <EOL> self . reader . setProtocol ( proto ) <EOL> return proto <EOL> def transferComplete ( self ) : <EOL> self . owner . transferComplete ( self ) <EOL> def __hash__ ( self ) : <EOL> return hash ( self . path ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . path ) <EOL> __repr__ = __str__ <EOL> def run_server ( ) : <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , help = '<STR_LIT>' , metavar = '<STR_LIT>' , default = '<STR_LIT:.>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , help = '<STR_LIT>' , <EOL> metavar = '<STR_LIT>' , default = <NUM_LIT> , type = int ) <EOL> parser . add_argument ( '<STR_LIT:-c>' , '<STR_LIT>' , type = int , default = <NUM_LIT:5> , metavar = '<STR_LIT>' , <EOL> help = """<STR_LIT>""" ) <EOL> parser . add_argument ( '<STR_LIT>' , type = int , default = <NUM_LIT:1> , metavar = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , help = '<STR_LIT>' , <EOL> action = '<STR_LIT:store_true>' , default = False ) <EOL> parser . add_argument ( '<STR_LIT>' , help = '<STR_LIT>' , action = '<STR_LIT:store_true>' , default = False ) <EOL> args = parser . parse_args ( ) <EOL> if args . debug : <EOL> global DaqRunner <EOL> DaqRunner = DummyDaqRunner <EOL> else : <EOL> if not DaqRunner : <EOL> raise __import_error <EOL> if args . verbose or args . debug : <EOL> log . start_logging ( '<STR_LIT>' ) <EOL> else : <EOL> log . start_logging ( '<STR_LIT>' ) <EOL> cleanup_period = args . cleanup_period * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> server = DaqServer ( args . directory ) <EOL> factory = DaqFactory ( server , cleanup_period , args . cleanup_after ) <EOL> reactor . listenTCP ( args . port , factory ) . getHost ( ) <EOL> try : <EOL> hostname = socket . gethostbyname ( socket . gethostname ( ) ) <EOL> except socket . gaierror : <EOL> hostname = '<STR_LIT:localhost>' <EOL> log . info ( '<STR_LIT>' . format ( hostname , args . port ) ) <EOL> reactor . run ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> run_server ( ) </s>
