<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> import threading <EOL> import logging <EOL> import string <EOL> import sys <EOL> import datetime <EOL> from openxc . formats . binary import ProtobufStreamer , ProtobufFormatter <EOL> from openxc . formats . json import JsonStreamer , JsonFormatter <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class MissingPayloadFormatError ( Exception ) : pass <EOL> class DataSource ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback = None , log_mode = None , payload_format = None ) : <EOL> """<STR_LIT>""" <EOL> super ( DataSource , self ) . __init__ ( ) <EOL> self . callback = callback <EOL> self . daemon = True <EOL> self . running = True <EOL> self . _streamer = None <EOL> self . _formatter = None <EOL> self . format = payload_format <EOL> self . logger = SourceLogger ( self , log_mode ) <EOL> @ property <EOL> def format ( self ) : <EOL> return self . _format <EOL> @ format . setter <EOL> def format ( self , value ) : <EOL> self . _format = value <EOL> if value == "<STR_LIT>" : <EOL> self . streamer = JsonStreamer ( ) <EOL> self . formatter = JsonFormatter <EOL> elif value == "<STR_LIT>" : <EOL> self . streamer = ProtobufStreamer ( ) <EOL> self . formatter = ProtobufFormatter <EOL> @ property <EOL> def streamer ( self ) : <EOL> if self . _streamer is None : <EOL> raise MissingPayloadFormatError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return self . _streamer <EOL> @ streamer . setter <EOL> def streamer ( self , value ) : <EOL> self . _streamer = value <EOL> @ property <EOL> def formatter ( self ) : <EOL> if self . _formatter is None : <EOL> raise MissingPayloadFormatError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return self . _formatter <EOL> @ formatter . setter <EOL> def formatter ( self , value ) : <EOL> self . _formatter = value <EOL> @ property <EOL> def bytes_received ( self ) : <EOL> return self . streamer . bytes_received <EOL> def start ( self ) : <EOL> self . logger . start ( ) <EOL> super ( DataSource , self ) . start ( ) <EOL> def stop ( self ) : <EOL> self . logger . stop ( ) <EOL> self . running = False <EOL> def read ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def read_logs ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> class SourceLogger ( threading . Thread ) : <EOL> FILENAME_TEMPLATE = "<STR_LIT>" <EOL> def __init__ ( self , source , mode = "<STR_LIT>" ) : <EOL> super ( SourceLogger , self ) . __init__ ( ) <EOL> self . daemon = True <EOL> self . source = source <EOL> self . mode = mode <EOL> self . file = None <EOL> self . running = True <EOL> if self . mode == "<STR_LIT:file>" : <EOL> filename = "<STR_LIT>" % datetime . datetime . now ( ) . strftime ( <EOL> self . FILENAME_TEMPLATE ) <EOL> self . file = open ( filename , '<STR_LIT>' ) <EOL> def stop ( self ) : <EOL> self . running = False <EOL> def record ( self , message ) : <EOL> if self . mode is not None and self . mode != "<STR_LIT>" and len ( message ) > <NUM_LIT:0> : <EOL> log_file = None <EOL> if self . mode == "<STR_LIT>" : <EOL> log_file = sys . stderr <EOL> elif self . mode == "<STR_LIT:file>" and self . file is not None : <EOL> log_file = self . file <EOL> print ( "<STR_LIT>" % message , file = log_file ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> message_buffer = b"<STR_LIT>" <EOL> while self . running : <EOL> try : <EOL> message_buffer += self . source . read_logs ( ) <EOL> except DataSourceError as e : <EOL> if self . running : <EOL> LOG . warn ( "<STR_LIT>" , e ) <EOL> break <EOL> except NotImplementedError as e : <EOL> LOG . info ( "<STR_LIT>" % self ) <EOL> break <EOL> while True : <EOL> if "<STR_LIT:\x00>" not in message_buffer : <EOL> break <EOL> record , _ , remainder = message_buffer . partition ( b"<STR_LIT:\x00>" ) <EOL> self . record ( record ) <EOL> message_buffer = remainder <EOL> class BytestreamDataSource ( DataSource ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( BytestreamDataSource , self ) . __init__ ( ** kwargs ) <EOL> self . corrupted_messages = <NUM_LIT:0> <EOL> self . running = True <EOL> def _message_valid ( self , message ) : <EOL> if not hasattr ( message , '<STR_LIT>' ) : <EOL> return False <EOL> if not ( '<STR_LIT:name>' in message and '<STR_LIT:value>' in message or <EOL> ( '<STR_LIT:id>' in message and '<STR_LIT:data>' in message ) or <EOL> ( '<STR_LIT:id>' in message and '<STR_LIT>' in message ) or <EOL> '<STR_LIT>' in message ) : <EOL> return False <EOL> return True <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . running : <EOL> try : <EOL> payload = self . read ( ) <EOL> except DataSourceError as e : <EOL> if self . running : <EOL> LOG . warn ( "<STR_LIT>" , e ) <EOL> break <EOL> try : <EOL> self . streamer <EOL> except MissingPayloadFormatError : <EOL> json_chars = [ '<STR_LIT:\x00>' ] <EOL> json_chars . extend ( string . printable ) <EOL> if all ( ( char in json_chars for char in payload ) ) : <EOL> self . format = "<STR_LIT>" <EOL> else : <EOL> self . format = "<STR_LIT>" <EOL> self . streamer . receive ( payload ) <EOL> while True : <EOL> message = self . streamer . parse_next_message ( ) <EOL> if message is None : <EOL> break <EOL> if not self . _message_valid ( message ) : <EOL> self . corrupted_messages += <NUM_LIT:1> <EOL> break <EOL> if self . callback is not None : <EOL> self . callback ( message ) <EOL> self . _receive_command_response ( message ) <EOL> def _receive_command_response ( self , message ) : <EOL> self . open_requests = getattr ( self , '<STR_LIT>' , [ ] ) <EOL> for open_request in self . open_requests : <EOL> open_request . put ( message ) <EOL> class DataSourceError ( Exception ) : <EOL> pass </s>
