<s> from __future__ import division , absolute_import <EOL> import functools <EOL> import platform <EOL> import six <EOL> import struct <EOL> import weakref <EOL> try : <EOL> from xcffib . _ffi import ffi <EOL> except ImportError : <EOL> from xcffib . ffi_build import ffi <EOL> SONAMES = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> try : <EOL> lib = ffi . dlopen ( SONAMES [ platform . system ( ) ] ) <EOL> except KeyError : <EOL> raise NotImplementedError ( "<STR_LIT>" % platform . system ( ) ) <EOL> __xcb_proto_version__ = '<STR_LIT>' <EOL> X_PROTOCOL = lib . X_PROTOCOL <EOL> X_PROTOCOL_REVISION = lib . X_PROTOCOL_REVISION <EOL> XCB_NONE = lib . XCB_NONE <EOL> XCB_COPY_FROM_PARENT = lib . XCB_COPY_FROM_PARENT <EOL> XCB_CURRENT_TIME = lib . XCB_CURRENT_TIME <EOL> XCB_NO_SYMBOL = lib . XCB_NO_SYMBOL <EOL> NONE = XCB_NONE <EOL> CopyFromParent = XCB_COPY_FROM_PARENT <EOL> CurrentTime = XCB_CURRENT_TIME <EOL> NoSymbol = XCB_NO_SYMBOL <EOL> XCB_CONN_ERROR = lib . XCB_CONN_ERROR <EOL> XCB_CONN_CLOSED_EXT_NOTSUPPORTED = lib . XCB_CONN_CLOSED_EXT_NOTSUPPORTED <EOL> XCB_CONN_CLOSED_MEM_INSUFFICIENT = lib . XCB_CONN_CLOSED_MEM_INSUFFICIENT <EOL> XCB_CONN_CLOSED_REQ_LEN_EXCEED = lib . XCB_CONN_CLOSED_REQ_LEN_EXCEED <EOL> XCB_CONN_CLOSED_PARSE_ERR = lib . XCB_CONN_CLOSED_PARSE_ERR <EOL> cffi_explicit_lifetimes = weakref . WeakKeyDictionary ( ) <EOL> def type_pad ( t , i ) : <EOL> return - i & ( <NUM_LIT:3> if t > <NUM_LIT:4> else t - <NUM_LIT:1> ) <EOL> def visualtype_to_c_struct ( vt ) : <EOL> s = ffi . new ( "<STR_LIT>" ) <EOL> s . visual_id = vt . visual_id <EOL> s . _class = vt . _class <EOL> s . bits_per_rgb_value = vt . bits_per_rgb_value <EOL> s . colormap_entries = vt . colormap_entries <EOL> s . red_mask = vt . red_mask <EOL> s . green_mask = vt . green_mask <EOL> s . blue_mask = vt . blue_mask <EOL> return s <EOL> class Unpacker ( object ) : <EOL> def __init__ ( self , known_max = None ) : <EOL> self . size = <NUM_LIT:0> <EOL> self . offset = <NUM_LIT:0> <EOL> self . known_max = known_max <EOL> if self . known_max is not None : <EOL> self . _resize ( known_max ) <EOL> def pad ( self , thing ) : <EOL> if isinstance ( thing , type ) and issubclass ( thing , ( Struct , Union ) ) : <EOL> if hasattr ( thing , "<STR_LIT>" ) : <EOL> size = thing . fixed_size <EOL> else : <EOL> size = <NUM_LIT:4> <EOL> else : <EOL> size = struct . calcsize ( thing ) <EOL> self . offset += type_pad ( size , self . offset ) <EOL> def unpack ( self , fmt , increment = True ) : <EOL> size = struct . calcsize ( fmt ) <EOL> if size > self . size - self . offset : <EOL> self . _resize ( size ) <EOL> ret = struct . unpack_from ( "<STR_LIT:=>" + fmt , self . buf , self . offset ) <EOL> if increment : <EOL> self . offset += size <EOL> return ret <EOL> def cast ( self , typ ) : <EOL> assert self . offset == <NUM_LIT:0> <EOL> return ffi . cast ( typ , self . cdata ) <EOL> def copy ( self ) : <EOL> raise NotImplementedError <EOL> @ classmethod <EOL> def synthetic ( cls , data , format ) : <EOL> self = cls . __new__ ( cls ) <EOL> self . buf = data <EOL> self . offset = <NUM_LIT:0> <EOL> self . size = len ( data ) <EOL> self . size <EOL> class CffiUnpacker ( Unpacker ) : <EOL> def __init__ ( self , cdata , known_max = None ) : <EOL> self . cdata = cdata <EOL> Unpacker . __init__ ( self , known_max ) <EOL> def _resize ( self , increment ) : <EOL> if self . offset + increment > self . size : <EOL> if self . known_max is not None : <EOL> assert self . size + increment <= self . known_max <EOL> self . size = self . offset + increment <EOL> self . buf = ffi . buffer ( self . cdata , self . size ) <EOL> def copy ( self ) : <EOL> new = CffiUnpacker ( self . cdata , self . known_max ) <EOL> new . offset = self . offset <EOL> new . size = self . size <EOL> return new <EOL> class MemoryUnpacker ( Unpacker ) : <EOL> def __init__ ( self , buf ) : <EOL> self . buf = buf <EOL> Unpacker . __init__ ( self , len ( self . buf ) ) <EOL> def _resize ( self , increment ) : <EOL> if self . size + increment > self . known_max : <EOL> raise XcffibException ( "<STR_LIT>" ) <EOL> self . size += increment <EOL> def copy ( self ) : <EOL> new = MemoryUnpacker ( self . buf ) <EOL> new . offset = self . offset <EOL> new . size = self . size <EOL> return new <EOL> def popcount ( n ) : <EOL> return bin ( n ) . count ( '<STR_LIT:1>' ) <EOL> class XcffibException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ConnectionException ( XcffibException ) : <EOL> REASONS = { <EOL> lib . XCB_CONN_ERROR : ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> lib . XCB_CONN_CLOSED_EXT_NOTSUPPORTED : ( <EOL> '<STR_LIT>' ) , <EOL> lib . XCB_CONN_CLOSED_MEM_INSUFFICIENT : ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> lib . XCB_CONN_CLOSED_REQ_LEN_EXCEED : ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> lib . XCB_CONN_CLOSED_PARSE_ERR : ( <EOL> '<STR_LIT>' ) , <EOL> } <EOL> def __init__ ( self , err ) : <EOL> XcffibException . __init__ ( <EOL> self , self . REASONS . get ( err , "<STR_LIT>" ) ) <EOL> class ProtocolException ( XcffibException ) : <EOL> pass <EOL> core = None <EOL> core_events = None <EOL> core_errors = None <EOL> _setup = None <EOL> extensions = { } <EOL> def _add_core ( value , __setup , events , errors ) : <EOL> if not issubclass ( value , Extension ) : <EOL> raise XcffibException ( <EOL> "<STR_LIT>" ) <EOL> if not issubclass ( __setup , Struct ) : <EOL> raise XcffibException ( "<STR_LIT>" ) <EOL> global core <EOL> global core_events <EOL> global core_errors <EOL> global _setup <EOL> core = value <EOL> core_events = events <EOL> core_errors = errors <EOL> _setup = __setup <EOL> def _add_ext ( key , value , events , errors ) : <EOL> if not issubclass ( value , Extension ) : <EOL> raise XcffibException ( <EOL> "<STR_LIT>" ) <EOL> extensions [ key ] = ( value , events , errors ) <EOL> class ExtensionKey ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> def __hash__ ( self ) : <EOL> return hash ( self . name ) <EOL> def __eq__ ( self , o ) : <EOL> return self . name == o . name <EOL> def __ne__ ( self , o ) : <EOL> return self . name != o . name <EOL> def to_cffi ( self ) : <EOL> c_key = ffi . new ( "<STR_LIT>" ) <EOL> c_key . name = name = ffi . new ( '<STR_LIT>' , self . name . encode ( ) ) <EOL> cffi_explicit_lifetimes [ c_key ] = name <EOL> c_key . global_id = <NUM_LIT:0> <EOL> return c_key <EOL> class Protobj ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , unpacker ) : <EOL> """<STR_LIT>""" <EOL> if unpacker . known_max is not None : <EOL> self . bufsize = unpacker . known_max <EOL> class Struct ( Protobj ) : <EOL> pass <EOL> class Union ( Protobj ) : <EOL> @ classmethod <EOL> def synthetic ( cls , data = [ ] , fmt = "<STR_LIT>" ) : <EOL> self = cls . __new__ ( cls ) <EOL> self . __init__ ( MemoryUnpacker ( struct . pack ( fmt , * data ) ) ) <EOL> return self <EOL> class Cookie ( object ) : <EOL> reply_type = None <EOL> def __init__ ( self , conn , sequence , is_checked ) : <EOL> self . conn = conn <EOL> self . sequence = sequence <EOL> self . is_checked = is_checked <EOL> def reply ( self ) : <EOL> data = self . conn . wait_for_reply ( self . sequence ) <EOL> return self . reply_type ( data ) <EOL> def check ( self ) : <EOL> assert self . is_checked and self . reply_type is None , ( <EOL> "<STR_LIT>" ) <EOL> self . conn . request_check ( self . sequence ) <EOL> class VoidCookie ( Cookie ) : <EOL> def reply ( self ) : <EOL> raise XcffibException ( "<STR_LIT>" ) <EOL> class Extension ( object ) : <EOL> def __init__ ( self , conn , key = None ) : <EOL> self . conn = conn <EOL> if key is None : <EOL> self . c_key = ffi . NULL <EOL> else : <EOL> c_key = key . to_cffi ( ) <EOL> cffi_explicit_lifetimes [ self ] = c_key <EOL> self . c_key = c_key <EOL> def send_request ( self , opcode , data , cookie = VoidCookie , reply = None , <EOL> is_checked = False ) : <EOL> data = data . getvalue ( ) <EOL> assert len ( data ) > <NUM_LIT:3> , "<STR_LIT>" <EOL> xcb_req = ffi . new ( "<STR_LIT>" ) <EOL> xcb_req . count = <NUM_LIT:2> <EOL> xcb_req . ext = self . c_key <EOL> xcb_req . opcode = opcode <EOL> xcb_req . isvoid = issubclass ( cookie , VoidCookie ) <EOL> xcb_parts = ffi . new ( "<STR_LIT>" ) <EOL> xcb_parts [ <NUM_LIT:2> ] . iov_base = iov_base = ffi . new ( '<STR_LIT>' , data ) <EOL> xcb_parts [ <NUM_LIT:2> ] . iov_len = len ( data ) <EOL> xcb_parts [ <NUM_LIT:3> ] . iov_base = ffi . NULL <EOL> xcb_parts [ <NUM_LIT:3> ] . iov_len = - len ( data ) & <NUM_LIT:3> <EOL> flags = lib . XCB_REQUEST_CHECKED if is_checked else <NUM_LIT:0> <EOL> seq = self . conn . send_request ( flags , xcb_parts + <NUM_LIT:2> , xcb_req ) <EOL> return cookie ( self . conn , seq , is_checked ) <EOL> def __getattr__ ( self , name ) : <EOL> if name . endswith ( "<STR_LIT>" ) : <EOL> real = name [ : - len ( "<STR_LIT>" ) ] <EOL> is_checked = True <EOL> elif name . endswith ( "<STR_LIT>" ) : <EOL> real = name [ : - len ( "<STR_LIT>" ) ] <EOL> is_checked = False <EOL> else : <EOL> raise AttributeError ( name ) <EOL> real = getattr ( self , real ) <EOL> return functools . partial ( real , is_checked = is_checked ) <EOL> class List ( Protobj ) : <EOL> def __init__ ( self , unpacker , typ , count = None ) : <EOL> Protobj . __init__ ( self , unpacker ) <EOL> self . list = [ ] <EOL> old = unpacker . offset <EOL> if isinstance ( typ , str ) : <EOL> self . list = list ( unpacker . unpack ( typ * count ) ) <EOL> elif count is not None : <EOL> for _ in range ( count ) : <EOL> item = typ ( unpacker ) <EOL> self . list . append ( item ) <EOL> else : <EOL> assert unpacker . known_max is not None <EOL> while unpacker . offset < unpacker . known_max : <EOL> item = typ ( unpacker ) <EOL> self . list . append ( item ) <EOL> self . bufsize = unpacker . offset - old <EOL> assert count is None or count == len ( self . list ) <EOL> def __str__ ( self ) : <EOL> return str ( self . list ) <EOL> def __len__ ( self ) : <EOL> return len ( self . list ) <EOL> def __iter__ ( self ) : <EOL> return iter ( self . list ) <EOL> def __getitem__ ( self , key ) : <EOL> return self . list [ key ] <EOL> def __setitem__ ( self , key , value ) : <EOL> self . list [ key ] = value <EOL> def __delitem__ ( self , key ) : <EOL> del self . list [ key ] <EOL> def to_string ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' . join ( chr ( six . byte2int ( i ) ) for i in self ) <EOL> def to_utf8 ( self ) : <EOL> return six . b ( '<STR_LIT>' ) . join ( self ) . decode ( '<STR_LIT:utf-8>' ) <EOL> def to_atoms ( self ) : <EOL> """<STR_LIT>""" <EOL> return struct . unpack ( "<STR_LIT:=>" + "<STR_LIT:I>" * ( len ( self ) // <NUM_LIT:4> ) , b'<STR_LIT>' . join ( self ) ) <EOL> def buf ( self ) : <EOL> return b'<STR_LIT>' . join ( self . list ) <EOL> @ classmethod <EOL> def synthetic ( cls , list = None ) : <EOL> if list is None : <EOL> list = [ ] <EOL> self = cls . __new__ ( cls ) <EOL> self . list = list [ : ] <EOL> return self <EOL> class OffsetMap ( object ) : <EOL> def __init__ ( self , core ) : <EOL> self . offsets = [ ( <NUM_LIT:0> , core ) ] <EOL> def add ( self , offset , things ) : <EOL> self . offsets . append ( ( offset , things ) ) <EOL> self . offsets . sort ( key = lambda x : x [ <NUM_LIT:0> ] , reverse = True ) <EOL> def __getitem__ ( self , item ) : <EOL> try : <EOL> offset , things = next ( ( k , v ) for k , v in self . offsets if item >= k ) <EOL> return things [ item - offset ] <EOL> except StopIteration : <EOL> raise IndexError ( item ) <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , display = None , fd = - <NUM_LIT:1> , auth = None ) : <EOL> if auth is not None : <EOL> [ name , data ] = auth . split ( six . b ( '<STR_LIT::>' ) ) <EOL> c_auth = ffi . new ( "<STR_LIT>" ) <EOL> c_auth . name = ffi . new ( '<STR_LIT>' , name ) <EOL> c_auth . namelen = len ( name ) <EOL> c_auth . data = ffi . new ( '<STR_LIT>' , data ) <EOL> c_auth . datalen = len ( data ) <EOL> else : <EOL> c_auth = ffi . NULL <EOL> if display is None : <EOL> display = ffi . NULL <EOL> else : <EOL> display = display . encode ( '<STR_LIT>' ) <EOL> i = ffi . new ( "<STR_LIT>" ) <EOL> if fd > <NUM_LIT:0> : <EOL> self . _conn = lib . xcb_connect_to_fd ( fd , c_auth ) <EOL> elif c_auth != ffi . NULL : <EOL> self . _conn = lib . xcb_connect_to_display_with_auth_info ( display , c_auth , i ) <EOL> else : <EOL> self . _conn = lib . xcb_connect ( display , i ) <EOL> self . pref_screen = i [ <NUM_LIT:0> ] <EOL> self . invalid ( ) <EOL> self . _init_x ( ) <EOL> def _init_x ( self ) : <EOL> if core is None : <EOL> raise XcffibException ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . core = core ( self ) <EOL> self . setup = self . get_setup ( ) <EOL> self . _event_offsets = OffsetMap ( core_events ) <EOL> self . _error_offsets = OffsetMap ( core_errors ) <EOL> self . _setup_extensions ( ) <EOL> def _setup_extensions ( self ) : <EOL> for key , ( _ , events , errors ) in extensions . items ( ) : <EOL> c_ext = key . to_cffi ( ) <EOL> reply = lib . xcb_get_extension_data ( self . _conn , c_ext ) <EOL> self . _event_offsets . add ( reply . first_event , events ) <EOL> self . _error_offsets . add ( reply . first_error , errors ) <EOL> def __call__ ( self , key ) : <EOL> return extensions [ key ] [ <NUM_LIT:0> ] ( self , key ) <EOL> def invalid ( self ) : <EOL> if self . _conn is None : <EOL> raise XcffibException ( "<STR_LIT>" ) <EOL> err = lib . xcb_connection_has_error ( self . _conn ) <EOL> if err > <NUM_LIT:0> : <EOL> raise ConnectionException ( err ) <EOL> def ensure_connected ( f ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( * args ) : <EOL> self = args [ <NUM_LIT:0> ] <EOL> self . invalid ( ) <EOL> try : <EOL> return f ( * args ) <EOL> finally : <EOL> self . invalid ( ) <EOL> return wrapper <EOL> @ ensure_connected <EOL> def get_setup ( self ) : <EOL> self . _setup = lib . xcb_get_setup ( self . _conn ) <EOL> buf = CffiUnpacker ( self . _setup , known_max = <NUM_LIT:8> + self . _setup . length * <NUM_LIT:4> ) <EOL> return _setup ( buf ) <EOL> @ ensure_connected <EOL> def get_screen_pointers ( self ) : <EOL> """<STR_LIT>""" <EOL> root_iter = lib . xcb_setup_roots_iterator ( self . _setup ) <EOL> screens = [ root_iter . data ] <EOL> for i in range ( self . _setup . roots_len - <NUM_LIT:1> ) : <EOL> lib . xcb_screen_next ( ffi . addressof ( ( root_iter ) ) ) <EOL> screens . append ( root_iter . data ) <EOL> return screens <EOL> @ ensure_connected <EOL> def wait_for_event ( self ) : <EOL> e = lib . xcb_wait_for_event ( self . _conn ) <EOL> e = ffi . gc ( e , lib . free ) <EOL> self . invalid ( ) <EOL> return self . hoist_event ( e ) <EOL> @ ensure_connected <EOL> def poll_for_event ( self ) : <EOL> e = lib . xcb_poll_for_event ( self . _conn ) <EOL> self . invalid ( ) <EOL> if e != ffi . NULL : <EOL> return self . hoist_event ( e ) <EOL> else : <EOL> return None <EOL> def has_error ( self ) : <EOL> return lib . xcb_connection_has_error ( self . _conn ) <EOL> @ ensure_connected <EOL> def get_file_descriptor ( self ) : <EOL> return lib . xcb_get_file_descriptor ( self . _conn ) <EOL> @ ensure_connected <EOL> def get_maximum_request_length ( self ) : <EOL> return lib . xcb_get_maximum_request_length ( self . _conn ) <EOL> @ ensure_connected <EOL> def prefetch_maximum_request_length ( self ) : <EOL> return lib . xcb_prefetch_maximum_request_length ( self . _conn ) <EOL> @ ensure_connected <EOL> def flush ( self ) : <EOL> return lib . xcb_flush ( self . _conn ) <EOL> @ ensure_connected <EOL> def generate_id ( self ) : <EOL> return lib . xcb_generate_id ( self . _conn ) <EOL> def disconnect ( self ) : <EOL> self . invalid ( ) <EOL> return lib . xcb_disconnect ( self . _conn ) <EOL> def _process_error ( self , c_error ) : <EOL> self . invalid ( ) <EOL> if c_error != ffi . NULL : <EOL> error = self . _error_offsets [ c_error . error_code ] <EOL> buf = CffiUnpacker ( c_error ) <EOL> raise error ( buf ) <EOL> @ ensure_connected <EOL> def wait_for_reply ( self , sequence ) : <EOL> error_p = ffi . new ( "<STR_LIT>" ) <EOL> data = lib . xcb_wait_for_reply ( self . _conn , sequence , error_p ) <EOL> data = ffi . gc ( data , lib . free ) <EOL> try : <EOL> self . _process_error ( error_p [ <NUM_LIT:0> ] ) <EOL> finally : <EOL> if error_p [ <NUM_LIT:0> ] != ffi . NULL : <EOL> lib . free ( error_p [ <NUM_LIT:0> ] ) <EOL> if data == ffi . NULL : <EOL> raise XcffibException ( "<STR_LIT>" % sequence ) <EOL> reply = ffi . cast ( "<STR_LIT>" , data ) <EOL> return CffiUnpacker ( data , known_max = <NUM_LIT:32> + reply . length * <NUM_LIT:4> ) <EOL> @ ensure_connected <EOL> def request_check ( self , sequence ) : <EOL> cookie = ffi . new ( "<STR_LIT>" ) <EOL> cookie [ <NUM_LIT:0> ] . sequence = sequence <EOL> err = lib . xcb_request_check ( self . _conn , cookie [ <NUM_LIT:0> ] ) <EOL> self . _process_error ( err ) <EOL> def hoist_event ( self , e ) : <EOL> """<STR_LIT>""" <EOL> if e . response_type == <NUM_LIT:0> : <EOL> return self . _process_error ( ffi . cast ( "<STR_LIT>" , e ) ) <EOL> event = self . _event_offsets [ e . response_type & <NUM_LIT> ] <EOL> buf = CffiUnpacker ( e ) <EOL> return event ( buf ) <EOL> @ ensure_connected <EOL> def send_request ( self , flags , xcb_parts , xcb_req ) : <EOL> return lib . xcb_send_request ( self . _conn , flags , xcb_parts , xcb_req ) <EOL> connect = Connection <EOL> class Response ( Protobj ) : <EOL> def __init__ ( self , unpacker ) : <EOL> Protobj . __init__ ( self , unpacker ) <EOL> if isinstance ( unpacker , CffiUnpacker ) : <EOL> resp = unpacker . cast ( "<STR_LIT>" ) <EOL> self . response_type = resp . response_type <EOL> self . sequence = resp . sequence <EOL> else : <EOL> self . response_type = None <EOL> self . sequence = None <EOL> class Reply ( Response ) : <EOL> def __init__ ( self , unpacker ) : <EOL> Response . __init__ ( self , unpacker ) <EOL> resp = unpacker . cast ( "<STR_LIT>" ) <EOL> self . length = resp . length <EOL> class Event ( Response ) : <EOL> pass <EOL> class Error ( Response , XcffibException ) : <EOL> def __init__ ( self , unpacker ) : <EOL> Response . __init__ ( self , unpacker ) <EOL> XcffibException . __init__ ( self ) <EOL> self . code = unpacker . unpack ( '<STR_LIT:B>' , increment = False ) <EOL> def pack_list ( from_ , pack_type ) : <EOL> """<STR_LIT>""" <EOL> if len ( from_ ) == <NUM_LIT:0> : <EOL> return bytes ( ) <EOL> if pack_type == '<STR_LIT:c>' : <EOL> if isinstance ( from_ , bytes ) : <EOL> from_ = [ six . int2byte ( b ) for b in six . iterbytes ( from_ ) ] <EOL> elif isinstance ( from_ , six . string_types ) : <EOL> from_ = [ six . int2byte ( b ) for b in bytearray ( from_ , '<STR_LIT:utf-8>' ) ] <EOL> elif isinstance ( from_ [ <NUM_LIT:0> ] , six . integer_types ) : <EOL> def to_bytes ( v ) : <EOL> for _ in range ( <NUM_LIT:4> ) : <EOL> v , r = divmod ( v , <NUM_LIT> ) <EOL> yield r <EOL> from_ = [ six . int2byte ( b ) for i in from_ for b in to_bytes ( i ) ] <EOL> if isinstance ( pack_type , six . string_types ) : <EOL> return struct . pack ( "<STR_LIT:=>" + pack_type * len ( from_ ) , * from_ ) <EOL> else : <EOL> buf = six . BytesIO ( ) <EOL> for item in from_ : <EOL> if isinstance ( item , Protobj ) and hasattr ( item , "<STR_LIT>" ) : <EOL> buf . write ( item . pack ( ) ) <EOL> else : <EOL> buf . write ( item ) <EOL> return buf . getvalue ( ) <EOL> def wrap ( ptr ) : <EOL> c_conn = ffi . cast ( '<STR_LIT>' , ptr ) <EOL> conn = Connection . __new__ ( Connection ) <EOL> conn . _conn = c_conn <EOL> conn . _init_x ( ) <EOL> conn . invalid ( ) <EOL> cffi_explicit_lifetimes [ conn ] = ptr <EOL> return conn </s>
