signature:
def charge_from_formula ( formula ) :
    r'''Basic formula parser to determine the charge from a formula - given
 that the charge is already specified as one element of the formula .
 Performs no sanity checking that elements are actually elements .
 Parameters
 formula : str
 Formula string , very simply formats only , ending in one of ' + x ' ,
 ' - x ' , n * ' + ' , or n * ' - ' or any of them surrounded by brackets but always
 at the end of a formula .
 Returns
 charge : int
 Charge of the molecule , [ faraday ]
 Notes
 Examples
 > > > charge _ from _ formula ( ' Br3 - ' )
 > > > charge _ from _ formula ( ' Br3 ( - ) ' )'''

implementation:
negative = '-' in formula
positive = '+' in formula
if positive and negative :
    raise ValueError ( 'Both negative and positive signs were found in the formula; only one sign is allowed' )
elif not ( positive or negative ) :
    return 0
multiplier , sign = ( - 1 , '-' ) if negative else ( 1 , '+' )
hit = False
if '(' in formula :
    hit = bracketed_charge_re . findall ( formula )
    if hit :
        formula = hit [ - 1 ] . replace ( '(' , '' ) . replace ( ')' , '' )
count = formula . count ( sign )
if count == 1 :
    splits = formula . split ( sign )
    if splits [ 1 ] == '' or splits [ 1 ] == ')' :
        return multiplier
    else :
        return multiplier * int ( splits [ 1 ] )
else :
    return multiplier * count


signature:
def serialize_formula ( formula ) :
    r'''Basic formula serializer to construct a consistently - formatted formula .
 This is necessary for handling user - supplied formulas , which are not always
 well formatted .
 Performs no sanity checking that elements are actually elements .
 Parameters
 formula : str
 Formula string as parseable by the method nested _ formula _ parser , [ - ]
 Returns
 formula : str
 A consistently formatted formula to describe a molecular formula , [ - ]
 Notes
 Examples
 > > > serialize _ formula ( ' Pd ( NH3)4 + 3 ' )
 ' H12N4Pd + 3' '''

implementation:
charge = charge_from_formula ( formula )
element_dict = nested_formula_parser ( formula )
base = atoms_to_Hill ( element_dict )
if charge == 0 :
    pass
elif charge > 0 :
    if charge == 1 :
        base += '+'
    else :
        base += '+' + str ( charge )
elif charge < 0 :
    if charge == - 1 :
        base += '-'
    else :
        base += str ( charge )
return base


signature:
async def connect ( self ) :
    """Establish a connection to the chat server .
 Returns when an error has occurred , or : func : ` disconnect ` has been
 called ."""

implementation:
proxy = os . environ . get ( 'HTTP_PROXY' )
self . _session = http_utils . Session ( self . _cookies , proxy = proxy )
try :
    self . _channel = channel . Channel ( self . _session , self . _max_retries , self . _retry_backoff_base )
    # Forward the Channel events to the Client events .
    self . _channel . on_connect . add_observer ( self . on_connect . fire )
    self . _channel . on_reconnect . add_observer ( self . on_reconnect . fire )
    self . _channel . on_disconnect . add_observer ( self . on_disconnect . fire )
    self . _channel . on_receive_array . add_observer ( self . _on_receive_array )
    # Wrap the coroutine in a Future so it can be cancelled .
    self . _listen_future = asyncio . ensure_future ( self . _channel . listen ( ) )
    # Listen for StateUpdate messages from the Channel until it
    # disconnects .
    try :
        await self . _listen_future
    except asyncio . CancelledError : # If this task is cancelled , we need to cancel our child task
    # as well . We don ' t need an additional yield because listen
    # cancels immediately .
        self . _listen_future . cancel ( )
    logger . info ( 'Client.connect returning because Channel.listen returned' )
finally :
    await self . _session . close ( )


signature:
def get_request_header ( self ) :
    """Return ` ` request _ header ` ` for use when constructing requests .
 Returns :
 Populated request header ."""

implementation:
# resource is allowed to be null if it ' s not available yet ( the Chrome
# client does this for the first getentitybyid call )
if self . _client_id is not None :
    self . _request_header . client_identifier . resource = self . _client_id
return self . _request_header


signature:
async def set_active ( self ) :
    """Set this client as active .
 While a client is active , no other clients will raise notifications .
 Call this method whenever there is an indication the user is
 interacting with this client . This method may be called very
 frequently , and it will only make a request when necessary ."""

implementation:
is_active = ( self . _active_client_state == hangouts_pb2 . ACTIVE_CLIENT_STATE_IS_ACTIVE )
timed_out = ( time . time ( ) - self . _last_active_secs > SETACTIVECLIENT_LIMIT_SECS )
if not is_active or timed_out : # Update these immediately so if the function is called again
# before the API request finishes , we don ' t start extra requests .
    self . _active_client_state = ( hangouts_pb2 . ACTIVE_CLIENT_STATE_IS_ACTIVE )
    self . _last_active_secs = time . time ( )
    # The first time this is called , we need to retrieve the user ' s
    # email address .
    if self . _email is None :
        try :
            get_self_info_request = hangouts_pb2 . GetSelfInfoRequest ( request_header = self . get_request_header ( ) , )
            get_self_info_response = await self . get_self_info ( get_self_info_request )
        except exceptions . NetworkError as e :
            logger . warning ( 'Failed to find email address: {}' . format ( e ) )
            return
        self . _email = ( get_self_info_response . self_entity . properties . email [ 0 ] )
    # If the client _ id hasn ' t been received yet , we can ' t set the
    # active client .
    if self . _client_id is None :
        logger . info ( 'Cannot set active client until client_id is received' )
        return
    try :
        set_active_request = hangouts_pb2 . SetActiveClientRequest ( request_header = self . get_request_header ( ) , is_active = True , full_jid = "{}/{}" . format ( self . _email , self . _client_id ) , timeout_secs = ACTIVE_TIMEOUT_SECS , )
        await self . set_active_client ( set_active_request )
    except exceptions . NetworkError as e :
        logger . warning ( 'Failed to set active client: {}' . format ( e ) )
    else :
        logger . info ( 'Set active client for {} seconds' . format ( ACTIVE_TIMEOUT_SECS ) )


signature:
async def upload_image ( self , image_file , filename = None , * , return_uploaded_image = False ) :
    """Upload an image that can be later attached to a chat message .
 Args :
 image _ file : A file - like object containing an image .
 filename ( str ) : ( optional ) Custom name for the uploaded file .
 return _ uploaded _ image ( bool ) : ( optional ) If True , return
 : class : ` . UploadedImage ` instead of image ID . Defaults to False .
 Raises :
 hangups . NetworkError : If the upload request failed .
 Returns :
 : class : ` . UploadedImage ` instance , or ID of the uploaded image ."""

implementation:
image_filename = filename or os . path . basename ( image_file . name )
image_data = image_file . read ( )
# request an upload URL
res = await self . _base_request ( IMAGE_UPLOAD_URL , 'application/x-www-form-urlencoded;charset=UTF-8' , 'json' , json . dumps ( { "protocolVersion" : "0.8" , "createSessionRequest" : { "fields" : [ { "external" : { "name" : "file" , "filename" : image_filename , "put" : { } , "size" : len ( image_data ) } } ] } } ) )
try :
    upload_url = self . _get_upload_session_status ( res ) [ 'externalFieldTransfers' ] [ 0 ] [ 'putInfo' ] [ 'url' ]
except KeyError :
    raise exceptions . NetworkError ( 'image upload failed: can not acquire an upload url' )
# upload the image data using the upload _ url to get the upload info
res = await self . _base_request ( upload_url , 'application/octet-stream' , 'json' , image_data )
try :
    raw_info = ( self . _get_upload_session_status ( res ) [ 'additionalInfo' ] [ 'uploader_service.GoogleRupioAdditionalInfo' ] [ 'completionInfo' ] [ 'customerSpecificInfo' ] )
    image_id = raw_info [ 'photoid' ]
    url = raw_info [ 'url' ]
except KeyError :
    raise exceptions . NetworkError ( 'image upload failed: can not fetch upload info' )
result = UploadedImage ( image_id = image_id , url = url )
return result if return_uploaded_image else result . image_id


signature:
def _get_upload_session_status ( res ) :
    """Parse the image upload response to obtain status .
 Args :
 res : http _ utils . FetchResponse instance , the upload response
 Returns :
 dict , sessionStatus of the response
 Raises :
 hangups . NetworkError : If the upload request failed ."""

implementation:
response = json . loads ( res . body . decode ( ) )
if 'sessionStatus' not in response :
    try :
        info = ( response [ 'errorMessage' ] [ 'additionalInfo' ] [ 'uploader_service.GoogleRupioAdditionalInfo' ] [ 'completionInfo' ] [ 'customerSpecificInfo' ] )
        reason = '{} : {}' . format ( info [ 'status' ] , info [ 'message' ] )
    except KeyError :
        reason = 'unknown reason'
    raise exceptions . NetworkError ( 'image upload failed: {}' . format ( reason ) )
return response [ 'sessionStatus' ]


signature:
async def _on_receive_array ( self , array ) :
    """Parse channel array and call the appropriate events ."""

implementation:
if array [ 0 ] == 'noop' :
    pass
    # This is just a keep - alive , ignore it .
else :
    wrapper = json . loads ( array [ 0 ] [ 'p' ] )
    # Wrapper appears to be a Protocol Buffer message , but encoded via
    # field numbers as dictionary keys . Since we don ' t have a parser
    # for that , parse it ad - hoc here .
    if '3' in wrapper : # This is a new client _ id .
        self . _client_id = wrapper [ '3' ] [ '2' ]
        logger . info ( 'Received new client_id: %r' , self . _client_id )
        # Once client _ id is received , the channel is ready to have
        # services added .
        await self . _add_channel_services ( )
    if '2' in wrapper :
        pblite_message = json . loads ( wrapper [ '2' ] [ '2' ] )
        if pblite_message [ 0 ] == 'cbu' : # This is a ( Client ) BatchUpdate containing StateUpdate
        # messages .
            batch_update = hangouts_pb2 . BatchUpdate ( )
            pblite . decode ( batch_update , pblite_message , ignore_first_item = True )
            for state_update in batch_update . state_update :
                logger . debug ( 'Received StateUpdate:\n%s' , state_update )
                header = state_update . state_update_header
                self . _active_client_state = header . active_client_state
                await self . on_state_update . fire ( state_update )
        else :
            logger . info ( 'Ignoring message: %r' , pblite_message [ 0 ] )


signature:
async def _add_channel_services ( self ) :
    """Add services to the channel .
 The services we add to the channel determine what kind of data we will
 receive on it .
 The " babel " service includes what we need for Hangouts . If this fails
 for some reason , hangups will never receive any events . The
 " babel _ presence _ last _ seen " service is also required to receive presence
 notifications .
 This needs to be re - called whenever we open a new channel ( when there ' s
 a new SID and client _ id ."""

implementation:
logger . info ( 'Adding channel services...' )
# Based on what Hangouts for Chrome does over 2 requests , this is
# trimmed down to 1 request that includes the bare minimum to make
# things work .
services = [ "babel" , "babel_presence_last_seen" ]
map_list = [ dict ( p = json . dumps ( { "3" : { "1" : { "1" : service } } } ) ) for service in services ]
await self . _channel . send_maps ( map_list )
logger . info ( 'Channel services added' )


signature:
async def _pb_request ( self , endpoint , request_pb , response_pb ) :
    """Send a Protocol Buffer formatted chat API request .
 Args :
 endpoint ( str ) : The chat API endpoint to use .
 request _ pb : The request body as a Protocol Buffer message .
 response _ pb : The response body as a Protocol Buffer message .
 Raises :
 NetworkError : If the request fails ."""

implementation:
logger . debug ( 'Sending Protocol Buffer request %s:\n%s' , endpoint , request_pb )
res = await self . _base_request ( 'https://clients6.google.com/chat/v1/{}' . format ( endpoint ) , 'application/x-protobuf' , # Request body is Protocol Buffer .
'proto' , # Response body is Protocol Buffer .
request_pb . SerializeToString ( ) )
try :
    response_pb . ParseFromString ( base64 . b64decode ( res . body ) )
except binascii . Error as e :
    raise exceptions . NetworkError ( 'Failed to decode base64 response: {}' . format ( e ) )
except google . protobuf . message . DecodeError as e :
    raise exceptions . NetworkError ( 'Failed to decode Protocol Buffer response: {}' . format ( e ) )
logger . debug ( 'Received Protocol Buffer response:\n%s' , response_pb )
status = response_pb . response_header . status
if status != hangouts_pb2 . RESPONSE_STATUS_OK :
    description = response_pb . response_header . error_description
    raise exceptions . NetworkError ( 'Request failed with status {}: \'{}\'' . format ( status , description ) )


signature:
async def _base_request ( self , url , content_type , response_type , data ) :
    """Send a generic authenticated POST request .
 Args :
 url ( str ) : URL of request .
 content _ type ( str ) : Request content type .
 response _ type ( str ) : The desired response format . Valid options
 are : ' json ' ( JSON ) , ' protojson ' ( pblite ) , and ' proto ' ( binary
 Protocol Buffer ) . ' proto ' requires manually setting an extra
 header ' X - Goog - Encode - Response - If - Executable : base64 ' .
 data ( str ) : Request body data .
 Returns :
 FetchResponse : Response containing HTTP code , cookies , and body .
 Raises :
 NetworkError : If the request fails ."""

implementation:
headers = { 'content-type' : content_type , # This header is required for Protocol Buffer responses . It causes
# them to be base64 encoded :
'X-Goog-Encode-Response-If-Executable' : 'base64' , }
params = { # " alternative representation type " ( desired response format ) .
'alt' : response_type , # API key ( required to avoid 403 Forbidden " Daily Limit for
# Unauthenticated Use Exceeded . Continued use requires signup " ) .
'key' : API_KEY , }
res = await self . _session . fetch ( 'post' , url , headers = headers , params = params , data = data , )
return res


signature:
async def add_user ( self , add_user_request ) :
    """Invite users to join an existing group conversation ."""

implementation:
response = hangouts_pb2 . AddUserResponse ( )
await self . _pb_request ( 'conversations/adduser' , add_user_request , response )
return response


signature:
async def create_conversation ( self , create_conversation_request ) :
    """Create a new conversation ."""

implementation:
response = hangouts_pb2 . CreateConversationResponse ( )
await self . _pb_request ( 'conversations/createconversation' , create_conversation_request , response )
return response


signature:
async def delete_conversation ( self , delete_conversation_request ) :
    """Leave a one - to - one conversation .
 One - to - one conversations are " sticky " ; they can ' t actually be deleted .
 This API clears the event history of the specified conversation up to
 ` ` delete _ upper _ bound _ timestamp ` ` , hiding it if no events remain ."""

implementation:
response = hangouts_pb2 . DeleteConversationResponse ( )
await self . _pb_request ( 'conversations/deleteconversation' , delete_conversation_request , response )
return response


signature:
async def easter_egg ( self , easter_egg_request ) :
    """Send an easter egg event to a conversation ."""

implementation:
response = hangouts_pb2 . EasterEggResponse ( )
await self . _pb_request ( 'conversations/easteregg' , easter_egg_request , response )
return response


signature:
async def get_conversation ( self , get_conversation_request ) :
    """Return conversation info and recent events ."""

implementation:
response = hangouts_pb2 . GetConversationResponse ( )
await self . _pb_request ( 'conversations/getconversation' , get_conversation_request , response )
return response


signature:
async def get_entity_by_id ( self , get_entity_by_id_request ) :
    """Return one or more user entities .
 Searching by phone number only finds entities when their phone number
 is in your contacts ( and not always even then ) , and can ' t be used to
 find Google Voice contacts ."""

implementation:
response = hangouts_pb2 . GetEntityByIdResponse ( )
await self . _pb_request ( 'contacts/getentitybyid' , get_entity_by_id_request , response )
return response


signature:
async def get_group_conversation_url ( self , get_group_conversation_url_request ) :
    """Get URL to allow others to join a group conversation ."""

implementation:
response = hangouts_pb2 . GetGroupConversationUrlResponse ( )
await self . _pb_request ( 'conversations/getgroupconversationurl' , get_group_conversation_url_request , response )
return response


signature:
async def get_self_info ( self , get_self_info_request ) :
    """Return info about the current user ."""

implementation:
response = hangouts_pb2 . GetSelfInfoResponse ( )
await self . _pb_request ( 'contacts/getselfinfo' , get_self_info_request , response )
return response


signature:
async def get_suggested_entities ( self , get_suggested_entities_request ) :
    """Return suggested contacts ."""

implementation:
response = hangouts_pb2 . GetSuggestedEntitiesResponse ( )
await self . _pb_request ( 'contacts/getsuggestedentities' , get_suggested_entities_request , response )
return response


signature:
async def query_presence ( self , query_presence_request ) :
    """Return presence status for a list of users ."""

implementation:
response = hangouts_pb2 . QueryPresenceResponse ( )
await self . _pb_request ( 'presence/querypresence' , query_presence_request , response )
return response


signature:
async def remove_user ( self , remove_user_request ) :
    """Remove a participant from a group conversation ."""

implementation:
response = hangouts_pb2 . RemoveUserResponse ( )
await self . _pb_request ( 'conversations/removeuser' , remove_user_request , response )
return response


signature:
async def rename_conversation ( self , rename_conversation_request ) :
    """Rename a conversation .
 Both group and one - to - one conversations may be renamed , but the
 official Hangouts clients have mixed support for one - to - one
 conversations with custom names ."""

implementation:
response = hangouts_pb2 . RenameConversationResponse ( )
await self . _pb_request ( 'conversations/renameconversation' , rename_conversation_request , response )
return response


signature:
async def search_entities ( self , search_entities_request ) :
    """Return user entities based on a query ."""

implementation:
response = hangouts_pb2 . SearchEntitiesResponse ( )
await self . _pb_request ( 'contacts/searchentities' , search_entities_request , response )
return response


signature:
async def send_chat_message ( self , send_chat_message_request ) :
    """Send a chat message to a conversation ."""

implementation:
response = hangouts_pb2 . SendChatMessageResponse ( )
await self . _pb_request ( 'conversations/sendchatmessage' , send_chat_message_request , response )
return response


signature:
async def modify_otr_status ( self , modify_otr_status_request ) :
    """Enable or disable message history in a conversation ."""

implementation:
response = hangouts_pb2 . ModifyOTRStatusResponse ( )
await self . _pb_request ( 'conversations/modifyotrstatus' , modify_otr_status_request , response )
return response


signature:
async def send_offnetwork_invitation ( self , send_offnetwork_invitation_request ) :
    """Send an email to invite a non - Google contact to Hangouts ."""

implementation:
response = hangouts_pb2 . SendOffnetworkInvitationResponse ( )
await self . _pb_request ( 'devices/sendoffnetworkinvitation' , send_offnetwork_invitation_request , response )
return response


signature:
async def set_active_client ( self , set_active_client_request ) :
    """Set the active client ."""

implementation:
response = hangouts_pb2 . SetActiveClientResponse ( )
await self . _pb_request ( 'clients/setactiveclient' , set_active_client_request , response )
return response


signature:
async def set_conversation_notification_level ( self , set_conversation_notification_level_request ) :
    """Set the notification level of a conversation ."""

implementation:
response = hangouts_pb2 . SetConversationNotificationLevelResponse ( )
await self . _pb_request ( 'conversations/setconversationnotificationlevel' , set_conversation_notification_level_request , response )
return response


signature:
async def set_focus ( self , set_focus_request ) :
    """Set focus to a conversation ."""

implementation:
response = hangouts_pb2 . SetFocusResponse ( )
await self . _pb_request ( 'conversations/setfocus' , set_focus_request , response )
return response


signature:
async def set_group_link_sharing_enabled ( self , set_group_link_sharing_enabled_request ) :
    """Set whether group link sharing is enabled for a conversation ."""

implementation:
response = hangouts_pb2 . SetGroupLinkSharingEnabledResponse ( )
await self . _pb_request ( 'conversations/setgrouplinksharingenabled' , set_group_link_sharing_enabled_request , response )
return response


signature:
async def set_presence ( self , set_presence_request ) :
    """Set the presence status ."""

implementation:
response = hangouts_pb2 . SetPresenceResponse ( )
await self . _pb_request ( 'presence/setpresence' , set_presence_request , response )
return response


signature:
async def set_typing ( self , set_typing_request ) :
    """Set the typing status of a conversation ."""

implementation:
response = hangouts_pb2 . SetTypingResponse ( )
await self . _pb_request ( 'conversations/settyping' , set_typing_request , response )
return response


signature:
async def sync_all_new_events ( self , sync_all_new_events_request ) :
    """List all events occurring at or after a timestamp ."""

implementation:
response = hangouts_pb2 . SyncAllNewEventsResponse ( )
await self . _pb_request ( 'conversations/syncallnewevents' , sync_all_new_events_request , response )
return response


signature:
async def sync_recent_conversations ( self , sync_recent_conversations_request ) :
    """Return info on recent conversations and their events ."""

implementation:
response = hangouts_pb2 . SyncRecentConversationsResponse ( )
await self . _pb_request ( 'conversations/syncrecentconversations' , sync_recent_conversations_request , response )
return response


signature:
async def update_watermark ( self , update_watermark_request ) :
    """Update the watermark ( read timestamp ) of a conversation ."""

implementation:
response = hangouts_pb2 . UpdateWatermarkResponse ( )
await self . _pb_request ( 'conversations/updatewatermark' , update_watermark_request , response )
return response


signature:
def from_timestamp ( microsecond_timestamp ) :
    """Convert a microsecond timestamp to a UTC datetime instance ."""

implementation:
# Create datetime without losing precision from floating point ( yes , this
# is actually needed ) :
return datetime . datetime . fromtimestamp ( microsecond_timestamp // 1000000 , datetime . timezone . utc ) . replace ( microsecond = ( microsecond_timestamp % 1000000 ) )


signature:
def from_participantid ( participant_id ) :
    """Convert hangouts _ pb2 . ParticipantId to UserID ."""

implementation: return user . UserID ( chat_id = participant_id . chat_id , gaia_id = participant_id . gaia_id )


signature:
def to_participantid ( user_id ) :
    """Convert UserID to hangouts _ pb2 . ParticipantId ."""

implementation: return hangouts_pb2 . ParticipantId ( chat_id = user_id . chat_id , gaia_id = user_id . gaia_id )


signature:
def parse_typing_status_message ( p ) :
    """Return TypingStatusMessage from hangouts _ pb2 . SetTypingNotification .
 The same status may be sent multiple times consecutively , and when a
 message is sent the typing status will not change to stopped ."""

implementation: return TypingStatusMessage ( conv_id = p . conversation_id . id , user_id = from_participantid ( p . sender_id ) , timestamp = from_timestamp ( p . timestamp ) , status = p . type , )


signature:
def parse_watermark_notification ( p ) :
    """Return WatermarkNotification from hangouts _ pb2 . WatermarkNotification ."""

implementation: return WatermarkNotification ( conv_id = p . conversation_id . id , user_id = from_participantid ( p . sender_id ) , read_timestamp = from_timestamp ( p . latest_read_timestamp ) , )


signature:
def _get_authorization_headers ( sapisid_cookie ) :
    """Return authorization headers for API request ."""

implementation:
# It doesn ' t seem to matter what the url and time are as long as they are
# consistent .
time_msec = int ( time . time ( ) * 1000 )
auth_string = '{} {} {}' . format ( time_msec , sapisid_cookie , ORIGIN_URL )
auth_hash = hashlib . sha1 ( auth_string . encode ( ) ) . hexdigest ( )
sapisidhash = 'SAPISIDHASH {}_{}' . format ( time_msec , auth_hash )
return { 'authorization' : sapisidhash , 'x-origin' : ORIGIN_URL , 'x-goog-authuser' : '0' , }


signature:
async def fetch ( self , method , url , params = None , headers = None , data = None ) :
    """Make an HTTP request .
 Automatically uses configured HTTP proxy , and adds Google authorization
 header and cookies .
 Failures will be retried MAX _ RETRIES times before raising NetworkError .
 Args :
 method ( str ) : Request method .
 url ( str ) : Request URL .
 params ( dict ) : ( optional ) Request query string parameters .
 headers ( dict ) : ( optional ) Request headers .
 data : ( str ) : ( optional ) Request body data .
 Returns :
 FetchResponse : Response data .
 Raises :
 NetworkError : If the request fails ."""

implementation:
logger . debug ( 'Sending request %s %s:\n%r' , method , url , data )
for retry_num in range ( MAX_RETRIES ) :
    try :
        async with self . fetch_raw ( method , url , params = params , headers = headers , data = data ) as res :
            async with async_timeout . timeout ( REQUEST_TIMEOUT ) :
                body = await res . read ( )
        logger . debug ( 'Received response %d %s:\n%r' , res . status , res . reason , body )
    except asyncio . TimeoutError :
        error_msg = 'Request timed out'
    except aiohttp . ServerDisconnectedError as err :
        error_msg = 'Server disconnected error: {}' . format ( err )
    except ( aiohttp . ClientError , ValueError ) as err :
        error_msg = 'Request connection error: {}' . format ( err )
    else :
        break
    logger . info ( 'Request attempt %d failed: %s' , retry_num , error_msg )
else :
    logger . info ( 'Request failed after %d attempts' , MAX_RETRIES )
    raise exceptions . NetworkError ( error_msg )
if res . status != 200 :
    logger . info ( 'Request returned unexpected status: %d %s' , res . status , res . reason )
    raise exceptions . NetworkError ( 'Request return unexpected status: {}: {}' . format ( res . status , res . reason ) )
return FetchResponse ( res . status , body )


signature:
def fetch_raw ( self , method , url , params = None , headers = None , data = None ) :
    """Make an HTTP request using aiohttp directly .
 Automatically uses configured HTTP proxy , and adds Google authorization
 header and cookies .
 Args :
 method ( str ) : Request method .
 url ( str ) : Request URL .
 params ( dict ) : ( optional ) Request query string parameters .
 headers ( dict ) : ( optional ) Request headers .
 data : ( str ) : ( optional ) Request body data .
 Returns :
 aiohttp . _ RequestContextManager : ContextManager for a HTTP response .
 Raises :
 See ` ` aiohttp . ClientSession . request ` ` ."""

implementation:
# Ensure we don ' t accidentally send the authorization header to a
# non - Google domain :
if not urllib . parse . urlparse ( url ) . hostname . endswith ( '.google.com' ) :
    raise Exception ( 'expected google.com domain' )
headers = headers or { }
headers . update ( self . _authorization_headers )
return self . _session . request ( method , url , params = params , headers = headers , data = data , proxy = self . _proxy )


signature:
async def lookup_entities ( client , args ) :
    """Search for entities by phone number , email , or gaia _ id ."""

implementation:
lookup_spec = _get_lookup_spec ( args . entity_identifier )
request = hangups . hangouts_pb2 . GetEntityByIdRequest ( request_header = client . get_request_header ( ) , batch_lookup_spec = [ lookup_spec ] , )
res = await client . get_entity_by_id ( request )
# Print the list of entities in the response .
for entity_result in res . entity_result :
    for entity in entity_result . entity :
        print ( entity )


signature:
def _get_lookup_spec ( identifier ) :
    """Return EntityLookupSpec from phone number , email address , or gaia ID ."""

implementation:
if identifier . startswith ( '+' ) :
    return hangups . hangouts_pb2 . EntityLookupSpec ( phone = identifier , create_offnetwork_gaia = True )
elif '@' in identifier :
    return hangups . hangouts_pb2 . EntityLookupSpec ( email = identifier , create_offnetwork_gaia = True )
else :
    return hangups . hangouts_pb2 . EntityLookupSpec ( gaia_id = identifier )


signature:
def get_conv_name ( conv , truncate = False , show_unread = False ) :
    """Return a readable name for a conversation .
 If the conversation has a custom name , use the custom name . Otherwise , for
 one - to - one conversations , the name is the full name of the other user . For
 group conversations , the name is a comma - separated list of first names . If
 the group conversation is empty , the name is " Empty Conversation " .
 If truncate is true , only show up to two names in a group conversation .
 If show _ unread is True , if there are unread chat messages , show the number
 of unread chat messages in parentheses after the conversation name ."""

implementation:
num_unread = len ( [ conv_event for conv_event in conv . unread_events if isinstance ( conv_event , hangups . ChatMessageEvent ) and not conv . get_user ( conv_event . user_id ) . is_self ] )
if show_unread and num_unread > 0 :
    postfix = ' ({})' . format ( num_unread )
else :
    postfix = ''
if conv . name is not None :
    return conv . name + postfix
else :
    participants = sorted ( ( user for user in conv . users if not user . is_self ) , key = lambda user : user . id_ )
    names = [ user . first_name for user in participants ]
    if not participants :
        return "Empty Conversation" + postfix
    if len ( participants ) == 1 :
        return participants [ 0 ] . full_name + postfix
    elif truncate and len ( participants ) > 2 :
        return ( ', ' . join ( names [ : 2 ] + [ '+{}' . format ( len ( names ) - 2 ) ] ) + postfix )
    else :
        return ', ' . join ( names ) + postfix


signature:
def add_color_to_scheme ( scheme , name , foreground , background , palette_colors ) :
    """Add foreground and background colours to a color scheme"""

implementation:
if foreground is None and background is None :
    return scheme
new_scheme = [ ]
for item in scheme :
    if item [ 0 ] == name :
        if foreground is None :
            foreground = item [ 1 ]
        if background is None :
            background = item [ 2 ]
        if palette_colors > 16 :
            new_scheme . append ( ( name , '' , '' , '' , foreground , background ) )
        else :
            new_scheme . append ( ( name , foreground , background ) )
    else :
        new_scheme . append ( item )
return new_scheme


signature:
async def build_user_conversation_list ( client ) :
    """Build : class : ` . UserList ` and : class : ` . ConversationList ` .
 This method requests data necessary to build the list of conversations and
 users . Users that are not in the contact list but are participating in a
 conversation will also be retrieved .
 Args :
 client ( Client ) : Connected client .
 Returns :
 ( : class : ` . UserList ` , : class : ` . ConversationList ` ) :
 Tuple of built objects ."""

implementation:
conv_states , sync_timestamp = await _sync_all_conversations ( client )
# Retrieve entities participating in all conversations .
required_user_ids = set ( )
for conv_state in conv_states :
    required_user_ids |= { user . UserID ( chat_id = part . id . chat_id , gaia_id = part . id . gaia_id ) for part in conv_state . conversation . participant_data }
required_entities = [ ]
if required_user_ids :
    logger . debug ( 'Need to request additional users: {}' . format ( required_user_ids ) )
    try :
        response = await client . get_entity_by_id ( hangouts_pb2 . GetEntityByIdRequest ( request_header = client . get_request_header ( ) , batch_lookup_spec = [ hangouts_pb2 . EntityLookupSpec ( gaia_id = user_id . gaia_id , create_offnetwork_gaia = True , ) for user_id in required_user_ids ] , ) )
        for entity_result in response . entity_result :
            required_entities . extend ( entity_result . entity )
    except exceptions . NetworkError as e :
        logger . warning ( 'Failed to request missing users: {}' . format ( e ) )
# Build list of conversation participants .
conv_part_list = [ ]
for conv_state in conv_states :
    conv_part_list . extend ( conv_state . conversation . participant_data )
# Retrieve self entity .
get_self_info_response = await client . get_self_info ( hangouts_pb2 . GetSelfInfoRequest ( request_header = client . get_request_header ( ) , ) )
self_entity = get_self_info_response . self_entity
user_list = user . UserList ( client , self_entity , required_entities , conv_part_list )
conversation_list = ConversationList ( client , conv_states , user_list , sync_timestamp )
return ( user_list , conversation_list )


signature:
async def _sync_all_conversations ( client ) :
    """Sync all conversations by making paginated requests .
 Conversations are ordered by ascending sort timestamp .
 Args :
 client ( Client ) : Connected client .
 Raises :
 NetworkError : If the requests fail .
 Returns :
 tuple of list of ` ` ConversationState ` ` messages and sync timestamp"""

implementation:
conv_states = [ ]
sync_timestamp = None
request = hangouts_pb2 . SyncRecentConversationsRequest ( request_header = client . get_request_header ( ) , max_conversations = CONVERSATIONS_PER_REQUEST , max_events_per_conversation = 1 , sync_filter = [ hangouts_pb2 . SYNC_FILTER_INBOX , hangouts_pb2 . SYNC_FILTER_ARCHIVED , ] )
for _ in range ( MAX_CONVERSATION_PAGES ) :
    logger . info ( 'Requesting conversations page %s' , request . last_event_timestamp )
    response = await client . sync_recent_conversations ( request )
    conv_states = list ( response . conversation_state ) + conv_states
    sync_timestamp = parsers . from_timestamp ( # SyncRecentConversations seems to return a sync _ timestamp 4
    # minutes before the present . To prevent SyncAllNewEvents later
    # breaking requesting events older than what we already have , use
    # current _ server _ time instead .
    response . response_header . current_server_time )
    if response . continuation_end_timestamp == 0 :
        logger . info ( 'Reached final conversations page' )
        break
    else :
        request . last_event_timestamp = response . continuation_end_timestamp
else :
    logger . warning ( 'Exceeded maximum number of conversation pages' )
logger . info ( 'Synced %s total conversations' , len ( conv_states ) )
return conv_states , sync_timestamp


signature:
def users ( self ) :
    """List of conversation participants ( : class : ` ~ hangups . user . User ` ) ."""

implementation: return [ self . _user_list . get_user ( user . UserID ( chat_id = part . id . chat_id , gaia_id = part . id . gaia_id ) ) for part in self . _conversation . participant_data ]


signature:
def last_modified ( self ) :
    """When conversation was last modified ( : class : ` datetime . datetime ` ) ."""

implementation:
timestamp = self . _conversation . self_conversation_state . sort_timestamp
# timestamp can be None for some reason when there is an ongoing video
# hangout
if timestamp is None :
    timestamp = 0
return parsers . from_timestamp ( timestamp )


signature:
def unread_events ( self ) :
    """Loaded events which are unread sorted oldest to newest .
 Some Hangouts clients don ' t update the read timestamp for certain event
 types , such as membership changes , so this may return more unread
 events than these clients will show . There ' s also a delay between
 sending a message and the user ' s own message being considered read .
 ( list of : class : ` . ConversationEvent ` ) ."""

implementation: return [ conv_event for conv_event in self . _events if conv_event . timestamp > self . latest_read_timestamp ]


signature:
def is_quiet ( self ) :
    """` ` True ` ` if notification level for this conversation is quiet ."""

implementation:
level = self . _conversation . self_conversation_state . notification_level
return level == hangouts_pb2 . NOTIFICATION_LEVEL_QUIET


signature:
def _on_watermark_notification ( self , notif ) :
    """Handle a watermark notification ."""

implementation:
# Update the conversation :
if self . get_user ( notif . user_id ) . is_self :
    logger . info ( 'latest_read_timestamp for {} updated to {}' . format ( self . id_ , notif . read_timestamp ) )
    self_conversation_state = ( self . _conversation . self_conversation_state )
    self_conversation_state . self_read_state . latest_read_timestamp = ( parsers . to_timestamp ( notif . read_timestamp ) )
# Update the participants ' watermarks :
previous_timestamp = self . _watermarks . get ( notif . user_id , datetime . datetime . min . replace ( tzinfo = datetime . timezone . utc ) )
if notif . read_timestamp > previous_timestamp :
    logger . info ( ( 'latest_read_timestamp for conv {} participant {}' + ' updated to {}' ) . format ( self . id_ , notif . user_id . chat_id , notif . read_timestamp ) )
    self . _watermarks [ notif . user_id ] = notif . read_timestamp


signature:
def update_conversation ( self , conversation ) :
    """Update the internal state of the conversation .
 This method is used by : class : ` . ConversationList ` to maintain this
 instance .
 Args :
 conversation : ` ` Conversation ` ` message ."""

implementation:
# StateUpdate . conversation is actually a delta ; fields that aren ' t
# specified are assumed to be unchanged . Until this class is
# refactored , hide this by saving and restoring previous values where
# necessary .
new_state = conversation . self_conversation_state
old_state = self . _conversation . self_conversation_state
self . _conversation = conversation
# delivery _ medium _ option
if not new_state . delivery_medium_option :
    new_state . delivery_medium_option . extend ( old_state . delivery_medium_option )
# latest _ read _ timestamp
old_timestamp = old_state . self_read_state . latest_read_timestamp
new_timestamp = new_state . self_read_state . latest_read_timestamp
if new_timestamp == 0 :
    new_state . self_read_state . latest_read_timestamp = old_timestamp
# user _ read _ state ( s )
for new_entry in conversation . read_state :
    tstamp = parsers . from_timestamp ( new_entry . latest_read_timestamp )
    if tstamp == 0 :
        continue
    uid = parsers . from_participantid ( new_entry . participant_id )
    if uid not in self . _watermarks or self . _watermarks [ uid ] < tstamp :
        self . _watermarks [ uid ] = tstamp


signature:
def _wrap_event ( event_ ) :
    """Wrap hangouts _ pb2 . Event in ConversationEvent subclass ."""

implementation:
cls = conversation_event . ConversationEvent
if event_ . HasField ( 'chat_message' ) :
    cls = conversation_event . ChatMessageEvent
elif event_ . HasField ( 'otr_modification' ) :
    cls = conversation_event . OTREvent
elif event_ . HasField ( 'conversation_rename' ) :
    cls = conversation_event . RenameEvent
elif event_ . HasField ( 'membership_change' ) :
    cls = conversation_event . MembershipChangeEvent
elif event_ . HasField ( 'hangout_event' ) :
    cls = conversation_event . HangoutEvent
elif event_ . HasField ( 'group_link_sharing_modification' ) :
    cls = conversation_event . GroupLinkSharingModificationEvent
return cls ( event_ )


signature:
def add_event ( self , event_ ) :
    """Add an event to the conversation .
 This method is used by : class : ` . ConversationList ` to maintain this
 instance .
 Args :
 event _ : ` ` Event ` ` message .
 Returns :
 : class : ` . ConversationEvent ` representing the event ."""

implementation:
conv_event = self . _wrap_event ( event_ )
if conv_event . id_ not in self . _events_dict :
    self . _events . append ( conv_event )
    self . _events_dict [ conv_event . id_ ] = conv_event
else : # If this happens , there ' s probably a bug .
    logger . info ( 'Conversation %s ignoring duplicate event %s' , self . id_ , conv_event . id_ )
    return None
return conv_event


signature:
def _get_default_delivery_medium ( self ) :
    """Return default DeliveryMedium to use for sending messages .
 Use the first option , or an option that ' s marked as the current
 default ."""

implementation:
medium_options = ( self . _conversation . self_conversation_state . delivery_medium_option )
try :
    default_medium = medium_options [ 0 ] . delivery_medium
except IndexError :
    logger . warning ( 'Conversation %r has no delivery medium' , self . id_ )
    default_medium = hangouts_pb2 . DeliveryMedium ( medium_type = hangouts_pb2 . DELIVERY_MEDIUM_BABEL )
for medium_option in medium_options :
    if medium_option . current_default :
        default_medium = medium_option . delivery_medium
return default_medium


signature:
def _get_event_request_header ( self ) :
    """Return EventRequestHeader for conversation ."""

implementation:
otr_status = ( hangouts_pb2 . OFF_THE_RECORD_STATUS_OFF_THE_RECORD if self . is_off_the_record else hangouts_pb2 . OFF_THE_RECORD_STATUS_ON_THE_RECORD )
return hangouts_pb2 . EventRequestHeader ( conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) , client_generated_id = self . _client . get_client_generated_id ( ) , expected_otr = otr_status , delivery_medium = self . _get_default_delivery_medium ( ) , )


signature:
async def send_message ( self , segments , image_file = None , image_id = None , image_user_id = None ) :
    """Send a message to this conversation .
 A per - conversation lock is acquired to ensure that messages are sent in
 the correct order when this method is called multiple times
 asynchronously .
 Args :
 segments : List of : class : ` . ChatMessageSegment ` objects to include
 in the message .
 image _ file : ( optional ) File - like object containing an image to be
 attached to the message .
 image _ id : ( optional ) ID of an Picasa photo to be attached to the
 message . If you specify both ` ` image _ file ` ` and ` ` image _ id ` `
 together , ` ` image _ file ` ` takes precedence and ` ` image _ id ` ` will
 be ignored .
 image _ user _ id : ( optional ) Picasa user ID , required only if
 ` ` image _ id ` ` refers to an image from a different Picasa user ,
 such as Google ' s sticker user .
 Raises :
 . NetworkError : If the message cannot be sent ."""

implementation:
async with self . _send_message_lock :
    if image_file :
        try :
            uploaded_image = await self . _client . upload_image ( image_file , return_uploaded_image = True )
        except exceptions . NetworkError as e :
            logger . warning ( 'Failed to upload image: {}' . format ( e ) )
            raise
        image_id = uploaded_image . image_id
    try :
        request = hangouts_pb2 . SendChatMessageRequest ( request_header = self . _client . get_request_header ( ) , event_request_header = self . _get_event_request_header ( ) , message_content = hangouts_pb2 . MessageContent ( segment = [ seg . serialize ( ) for seg in segments ] , ) , )
        if image_id is not None :
            request . existing_media . photo . photo_id = image_id
        if image_user_id is not None :
            request . existing_media . photo . user_id = image_user_id
            request . existing_media . photo . is_custom_user_id = True
        await self . _client . send_chat_message ( request )
    except exceptions . NetworkError as e :
        logger . warning ( 'Failed to send message: {}' . format ( e ) )
        raise


signature:
async def leave ( self ) :
    """Leave this conversation .
 Raises :
 . NetworkError : If conversation cannot be left ."""

implementation:
is_group_conversation = ( self . _conversation . type == hangouts_pb2 . CONVERSATION_TYPE_GROUP )
try :
    if is_group_conversation :
        await self . _client . remove_user ( hangouts_pb2 . RemoveUserRequest ( request_header = self . _client . get_request_header ( ) , event_request_header = self . _get_event_request_header ( ) , ) )
    else :
        await self . _client . delete_conversation ( hangouts_pb2 . DeleteConversationRequest ( request_header = self . _client . get_request_header ( ) , conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) , delete_upper_bound_timestamp = parsers . to_timestamp ( datetime . datetime . now ( tz = datetime . timezone . utc ) ) ) )
except exceptions . NetworkError as e :
    logger . warning ( 'Failed to leave conversation: {}' . format ( e ) )
    raise


signature:
async def rename ( self , name ) :
    """Rename this conversation .
 Hangouts only officially supports renaming group conversations , so
 custom names for one - to - one conversations may or may not appear in all
 first party clients .
 Args :
 name ( str ) : New name .
 Raises :
 . NetworkError : If conversation cannot be renamed ."""

implementation: await self . _client . rename_conversation ( hangouts_pb2 . RenameConversationRequest ( request_header = self . _client . get_request_header ( ) , new_name = name , event_request_header = self . _get_event_request_header ( ) , ) )


signature:
async def set_notification_level ( self , level ) :
    """Set the notification level of this conversation .
 Args :
 level : ` ` NOTIFICATION _ LEVEL _ QUIET ` ` to disable notifications , or
 ` ` NOTIFICATION _ LEVEL _ RING ` ` to enable them .
 Raises :
 . NetworkError : If the request fails ."""

implementation: await self . _client . set_conversation_notification_level ( hangouts_pb2 . SetConversationNotificationLevelRequest ( request_header = self . _client . get_request_header ( ) , conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) , level = level , ) )


signature:
async def set_typing ( self , typing = hangouts_pb2 . TYPING_TYPE_STARTED ) :
    """Set your typing status in this conversation .
 Args :
 typing : ( optional ) ` ` TYPING _ TYPE _ STARTED ` ` , ` ` TYPING _ TYPE _ PAUSED ` ` ,
 or ` ` TYPING _ TYPE _ STOPPED ` ` to start , pause , or stop typing ,
 respectively . Defaults to ` ` TYPING _ TYPE _ STARTED ` ` .
 Raises :
 . NetworkError : If typing status cannot be set ."""

implementation:
# TODO : Add rate - limiting to avoid unnecessary requests .
try :
    await self . _client . set_typing ( hangouts_pb2 . SetTypingRequest ( request_header = self . _client . get_request_header ( ) , conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) , type = typing , ) )
except exceptions . NetworkError as e :
    logger . warning ( 'Failed to set typing status: {}' . format ( e ) )
    raise


signature:
async def update_read_timestamp ( self , read_timestamp = None ) :
    """Update the timestamp of the latest event which has been read .
 This method will avoid making an API request if it will have no effect .
 Args :
 read _ timestamp ( datetime . datetime ) : ( optional ) Timestamp to set .
 Defaults to the timestamp of the newest event .
 Raises :
 . NetworkError : If the timestamp cannot be updated ."""

implementation:
if read_timestamp is None :
    read_timestamp = ( self . events [ - 1 ] . timestamp if self . events else datetime . datetime . now ( datetime . timezone . utc ) )
if read_timestamp > self . latest_read_timestamp :
    logger . info ( 'Setting {} latest_read_timestamp from {} to {}' . format ( self . id_ , self . latest_read_timestamp , read_timestamp ) )
    # Prevent duplicate requests by updating the conversation now .
    state = self . _conversation . self_conversation_state
    state . self_read_state . latest_read_timestamp = ( parsers . to_timestamp ( read_timestamp ) )
    try :
        await self . _client . update_watermark ( hangouts_pb2 . UpdateWatermarkRequest ( request_header = self . _client . get_request_header ( ) , conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) , last_read_timestamp = parsers . to_timestamp ( read_timestamp ) , ) )
    except exceptions . NetworkError as e :
        logger . warning ( 'Failed to update read timestamp: {}' . format ( e ) )
        raise


signature:
async def get_events ( self , event_id = None , max_events = 50 ) :
    """Get events from this conversation .
 Makes a request to load historical events if necessary .
 Args :
 event _ id ( str ) : ( optional ) If provided , return events preceding
 this event , otherwise return the newest events .
 max _ events ( int ) : Maximum number of events to return . Defaults to
 50.
 Returns :
 List of : class : ` . ConversationEvent ` instances , ordered
 newest - first .
 Raises :
 KeyError : If ` ` event _ id ` ` does not correspond to a known event .
 . NetworkError : If the events could not be requested ."""

implementation:
if event_id is None : # If no event _ id is provided , return the newest events in this
# conversation .
    conv_events = self . _events [ - 1 * max_events : ]
else : # If event _ id is provided , return the events we have that are
# older , or request older events if event _ id corresponds to the
# oldest event we have .
    conv_event = self . get_event ( event_id )
    if self . _events [ 0 ] . id_ != event_id :
        conv_events = self . _events [ self . _events . index ( conv_event ) + 1 : ]
    else :
        logger . info ( 'Loading events for conversation {} before {}' . format ( self . id_ , conv_event . timestamp ) )
        res = await self . _client . get_conversation ( hangouts_pb2 . GetConversationRequest ( request_header = self . _client . get_request_header ( ) , conversation_spec = hangouts_pb2 . ConversationSpec ( conversation_id = hangouts_pb2 . ConversationId ( id = self . id_ ) ) , include_event = True , max_events_per_conversation = max_events , event_continuation_token = self . _event_cont_token ) )
        # Certain fields of conversation _ state are not populated by
        # SyncRecentConversations . This is the case with the
        # user _ read _ state fields which are all set to 0 but for the
        # ' self ' user . Update here so these fields get populated on the
        # first call to GetConversation .
        if res . conversation_state . HasField ( 'conversation' ) :
            self . update_conversation ( res . conversation_state . conversation )
        self . _event_cont_token = ( res . conversation_state . event_continuation_token )
        conv_events = [ self . _wrap_event ( event ) for event in res . conversation_state . event ]
        logger . info ( 'Loaded {} events for conversation {}' . format ( len ( conv_events ) , self . id_ ) )
        # Iterate though the events newest to oldest .
        for conv_event in reversed ( conv_events ) : # Add event as the new oldest event , unless we already have
        # it .
            if conv_event . id_ not in self . _events_dict :
                self . _events . insert ( 0 , conv_event )
                self . _events_dict [ conv_event . id_ ] = conv_event
            else : # If this happens , there ' s probably a bug .
                logger . info ( 'Conversation %s ignoring duplicate event %s' , self . id_ , conv_event . id_ )
return conv_events


signature:
def next_event ( self , event_id , prev = False ) :
    """Get the event following another event in this conversation .
 Args :
 event _ id ( str ) : ID of the event .
 prev ( bool ) : If ` ` True ` ` , return the previous event rather than the
 next event . Defaults to ` ` False ` ` .
 Raises :
 KeyError : If no such : class : ` . ConversationEvent ` is known .
 Returns :
 : class : ` . ConversationEvent ` or ` ` None ` ` if there is no following
 event ."""

implementation:
i = self . events . index ( self . _events_dict [ event_id ] )
if prev and i > 0 :
    return self . events [ i - 1 ]
elif not prev and i + 1 < len ( self . events ) :
    return self . events [ i + 1 ]
else :
    return None


signature:
def get_all ( self , include_archived = False ) :
    """Get all the conversations .
 Args :
 include _ archived ( bool ) : ( optional ) Whether to include archived
 conversations . Defaults to ` ` False ` ` .
 Returns :
 List of all : class : ` . Conversation ` objects ."""

implementation: return [ conv for conv in self . _conv_dict . values ( ) if not conv . is_archived or include_archived ]


signature:
async def leave_conversation ( self , conv_id ) :
    """Leave a conversation .
 Args :
 conv _ id ( str ) : ID of conversation to leave ."""

implementation:
logger . info ( 'Leaving conversation: {}' . format ( conv_id ) )
await self . _conv_dict [ conv_id ] . leave ( )
del self . _conv_dict [ conv_id ]


signature:
def _add_conversation ( self , conversation , events = [ ] , event_cont_token = None ) :
    """Add new conversation from hangouts _ pb2 . Conversation"""

implementation:
# pylint : disable = dangerous - default - value
conv_id = conversation . conversation_id . id
logger . debug ( 'Adding new conversation: {}' . format ( conv_id ) )
conv = Conversation ( self . _client , self . _user_list , conversation , events , event_cont_token )
self . _conv_dict [ conv_id ] = conv
return conv


signature:
async def _on_state_update ( self , state_update ) :
    """Receive a StateUpdate and fan out to Conversations .
 Args :
 state _ update : hangouts _ pb2 . StateUpdate instance"""

implementation:
# The state update will include some type of notification :
notification_type = state_update . WhichOneof ( 'state_update' )
# If conversation fields have been updated , the state update will have
# a conversation containing changed fields . Handle updating the
# conversation from this delta :
if state_update . HasField ( 'conversation' ) :
    try :
        await self . _handle_conversation_delta ( state_update . conversation )
    except exceptions . NetworkError :
        logger . warning ( 'Discarding %s for %s: Failed to fetch conversation' , notification_type . replace ( '_' , ' ' ) , state_update . conversation . conversation_id . id )
        return
if notification_type == 'typing_notification' :
    await self . _handle_set_typing_notification ( state_update . typing_notification )
elif notification_type == 'watermark_notification' :
    await self . _handle_watermark_notification ( state_update . watermark_notification )
elif notification_type == 'event_notification' :
    await self . _on_event ( state_update . event_notification . event )


signature:
async def _get_or_fetch_conversation ( self , conv_id ) :
    """Get a cached conversation or fetch a missing conversation .
 Args :
 conv _ id : string , conversation identifier
 Raises :
 NetworkError : If the request to fetch the conversation fails .
 Returns :
 : class : ` . Conversation ` with matching ID ."""

implementation:
conv = self . _conv_dict . get ( conv_id , None )
if conv is None :
    logger . info ( 'Fetching unknown conversation %s' , conv_id )
    res = await self . _client . get_conversation ( hangouts_pb2 . GetConversationRequest ( request_header = self . _client . get_request_header ( ) , conversation_spec = hangouts_pb2 . ConversationSpec ( conversation_id = hangouts_pb2 . ConversationId ( id = conv_id ) ) , include_event = False ) )
    conv_state = res . conversation_state
    event_cont_token = None
    if conv_state . HasField ( 'event_continuation_token' ) :
        event_cont_token = conv_state . event_continuation_token
    return self . _add_conversation ( conv_state . conversation , event_cont_token = event_cont_token )
else :
    return conv


signature:
async def _on_event ( self , event_ ) :
    """Receive a hangouts _ pb2 . Event and fan out to Conversations .
 Args :
 event _ : hangouts _ pb2 . Event instance"""

implementation:
conv_id = event_ . conversation_id . id
try :
    conv = await self . _get_or_fetch_conversation ( conv_id )
except exceptions . NetworkError :
    logger . warning ( 'Failed to fetch conversation for event notification: %s' , conv_id )
else :
    self . _sync_timestamp = parsers . from_timestamp ( event_ . timestamp )
    conv_event = conv . add_event ( event_ )
    # conv _ event may be None if the event was a duplicate .
    if conv_event is not None :
        await self . on_event . fire ( conv_event )
        await conv . on_event . fire ( conv_event )


signature:
async def _handle_conversation_delta ( self , conversation ) :
    """Receive Conversation delta and create or update the conversation .
 Args :
 conversation : hangouts _ pb2 . Conversation instance
 Raises :
 NetworkError : A request to fetch the complete conversation failed ."""

implementation:
conv_id = conversation . conversation_id . id
conv = self . _conv_dict . get ( conv_id , None )
if conv is None : # Ignore the delta and fetch the complete conversation .
    await self . _get_or_fetch_conversation ( conv_id )
else : # Update conversation using the delta .
    conv . update_conversation ( conversation )


signature:
async def _handle_set_typing_notification ( self , set_typing_notification ) :
    """Receive SetTypingNotification and update the conversation .
 Args :
 set _ typing _ notification : hangouts _ pb2 . SetTypingNotification
 instance"""

implementation:
conv_id = set_typing_notification . conversation_id . id
res = parsers . parse_typing_status_message ( set_typing_notification )
await self . on_typing . fire ( res )
try :
    conv = await self . _get_or_fetch_conversation ( conv_id )
except exceptions . NetworkError :
    logger . warning ( 'Failed to fetch conversation for typing notification: %s' , conv_id )
else :
    await conv . on_typing . fire ( res )


signature:
async def _handle_watermark_notification ( self , watermark_notification ) :
    """Receive WatermarkNotification and update the conversation .
 Args :
 watermark _ notification : hangouts _ pb2 . WatermarkNotification instance"""

implementation:
conv_id = watermark_notification . conversation_id . id
res = parsers . parse_watermark_notification ( watermark_notification )
await self . on_watermark_notification . fire ( res )
try :
    conv = await self . _get_or_fetch_conversation ( conv_id )
except exceptions . NetworkError :
    logger . warning ( 'Failed to fetch conversation for watermark notification: %s' , conv_id )
else :
    await conv . on_watermark_notification . fire ( res )


signature:
async def _sync ( self ) :
    """Sync conversation state and events that could have been missed ."""

implementation:
logger . info ( 'Syncing events since {}' . format ( self . _sync_timestamp ) )
try :
    res = await self . _client . sync_all_new_events ( hangouts_pb2 . SyncAllNewEventsRequest ( request_header = self . _client . get_request_header ( ) , last_sync_timestamp = parsers . to_timestamp ( self . _sync_timestamp ) , max_response_size_bytes = 1048576 , # 1 MB
    ) )
except exceptions . NetworkError as e :
    logger . warning ( 'Failed to sync events, some events may be lost: {}' . format ( e ) )
else :
    for conv_state in res . conversation_state :
        conv_id = conv_state . conversation_id . id
        conv = self . _conv_dict . get ( conv_id , None )
        if conv is not None :
            conv . update_conversation ( conv_state . conversation )
            for event_ in conv_state . event :
                timestamp = parsers . from_timestamp ( event_ . timestamp )
                if timestamp > self . _sync_timestamp : # This updates the sync _ timestamp for us , as well
                # as triggering events .
                    await self . _on_event ( event_ )
        else :
            self . _add_conversation ( conv_state . conversation , conv_state . event , conv_state . event_continuation_token )


signature:
def upgrade_name ( self , user_ ) :
    """Upgrade name type of this user .
 Google Voice participants often first appear with no name at all , and
 then get upgraded unpredictably to numbers ( " + 12125551212 " ) or names .
 Args :
 user _ ( ~ hangups . user . User ) : User to upgrade with ."""

implementation:
if user_ . name_type > self . name_type :
    self . full_name = user_ . full_name
    self . first_name = user_ . first_name
    self . name_type = user_ . name_type
    logger . debug ( 'Added %s name to User "%s": %s' , self . name_type . name . lower ( ) , self . full_name , self )


signature:
def from_entity ( entity , self_user_id ) :
    """Construct user from ` ` Entity ` ` message .
 Args :
 entity : ` ` Entity ` ` message .
 self _ user _ id ( ~ hangups . user . UserID or None ) : The ID of the current
 user . If ` ` None ` ` , assume ` ` entity ` ` is the current user .
 Returns :
 : class : ` ~ hangups . user . User ` object ."""

implementation:
user_id = UserID ( chat_id = entity . id . chat_id , gaia_id = entity . id . gaia_id )
return User ( user_id , entity . properties . display_name , entity . properties . first_name , entity . properties . photo_url , entity . properties . email , ( self_user_id == user_id ) or ( self_user_id is None ) )


signature:
def from_conv_part_data ( conv_part_data , self_user_id ) :
    """Construct user from ` ` ConversationParticipantData ` ` message .
 Args :
 conv _ part _ id : ` ` ConversationParticipantData ` ` message .
 self _ user _ id ( ~ hangups . user . UserID or None ) : The ID of the current
 user . If ` ` None ` ` , assume ` ` conv _ part _ id ` ` is the current user .
 Returns :
 : class : ` ~ hangups . user . User ` object ."""

implementation:
user_id = UserID ( chat_id = conv_part_data . id . chat_id , gaia_id = conv_part_data . id . gaia_id )
return User ( user_id , conv_part_data . fallback_name , None , None , [ ] , ( self_user_id == user_id ) or ( self_user_id is None ) )


signature:
def get_user ( self , user_id ) :
    """Get a user by its ID .
 Args :
 user _ id ( ~ hangups . user . UserID ) : The ID of the user .
 Raises :
 KeyError : If no such user is known .
 Returns :
 : class : ` ~ hangups . user . User ` with the given ID ."""

implementation:
try :
    return self . _user_dict [ user_id ]
except KeyError :
    logger . warning ( 'UserList returning unknown User for UserID %s' , user_id )
    return User ( user_id , None , None , None , [ ] , False )


signature:
def _add_user_from_conv_part ( self , conv_part ) :
    """Add or upgrade User from ConversationParticipantData ."""

implementation:
user_ = User . from_conv_part_data ( conv_part , self . _self_user . id_ )
existing = self . _user_dict . get ( user_ . id_ )
if existing is None :
    logger . warning ( 'Adding fallback User with %s name "%s"' , user_ . name_type . name . lower ( ) , user_ . full_name )
    self . _user_dict [ user_ . id_ ] = user_
    return user_
else :
    existing . upgrade_name ( user_ )
    return existing


signature:
def add_observer ( self , callback ) :
    """Add an observer to this event .
 Args :
 callback : A function or coroutine callback to call when the event
 is fired .
 Raises :
 ValueError : If the callback has already been added ."""

implementation:
if callback in self . _observers :
    raise ValueError ( '{} is already an observer of {}' . format ( callback , self ) )
self . _observers . append ( callback )


signature:
def remove_observer ( self , callback ) :
    """Remove an observer from this event .
 Args :
 callback : A function or coroutine callback to remove from this
 event .
 Raises :
 ValueError : If the callback is not an observer of this event ."""

implementation:
if callback not in self . _observers :
    raise ValueError ( '{} is not an observer of {}' . format ( callback , self ) )
self . _observers . remove ( callback )


signature:
async def fire ( self , * args , ** kwargs ) :
    """Fire this event , calling all observers with the same arguments ."""

implementation:
logger . debug ( 'Fired {}' . format ( self ) )
for observer in self . _observers :
    gen = observer ( * args , ** kwargs )
    if asyncio . iscoroutinefunction ( observer ) :
        await gen


signature:
def markdown ( tag ) :
    """Return start and end regex pattern sequences for simple Markdown tag ."""

implementation: return ( MARKDOWN_START . format ( tag = tag ) , MARKDOWN_END . format ( tag = tag ) )


signature:
def html ( tag ) :
    """Return sequence of start and end regex patterns for simple HTML tag"""

implementation: return ( HTML_START . format ( tag = tag ) , HTML_END . format ( tag = tag ) )


signature:
def run_example ( example_coroutine , * extra_args ) :
    """Run a hangups example coroutine .
 Args :
 example _ coroutine ( coroutine ) : Coroutine to run with a connected
 hangups client and arguments namespace as arguments .
 extra _ args ( str ) : Any extra command line arguments required by the
 example ."""

implementation:
args = _get_parser ( extra_args ) . parse_args ( )
logging . basicConfig ( level = logging . DEBUG if args . debug else logging . WARNING )
# Obtain hangups authentication cookies , prompting for credentials from
# standard input if necessary .
cookies = hangups . auth . get_auth_stdin ( args . token_path )
client = hangups . Client ( cookies )
loop = asyncio . get_event_loop ( )
task = asyncio . ensure_future ( _async_main ( example_coroutine , client , args ) , loop = loop )
try :
    loop . run_until_complete ( task )
except KeyboardInterrupt :
    task . cancel ( )
    loop . run_until_complete ( task )
finally :
    loop . close ( )


signature:
def _get_parser ( extra_args ) :
    """Return ArgumentParser with any extra arguments ."""

implementation:
parser = argparse . ArgumentParser ( formatter_class = argparse . ArgumentDefaultsHelpFormatter , )
dirs = appdirs . AppDirs ( 'hangups' , 'hangups' )
default_token_path = os . path . join ( dirs . user_cache_dir , 'refresh_token.txt' )
parser . add_argument ( '--token-path' , default = default_token_path , help = 'path used to store OAuth refresh token' )
parser . add_argument ( '-d' , '--debug' , action = 'store_true' , help = 'log detailed debugging messages' )
for extra_arg in extra_args :
    parser . add_argument ( extra_arg , required = True )
return parser


signature:
async def _async_main ( example_coroutine , client , args ) :
    """Run the example coroutine ."""

implementation:
# Spawn a task for hangups to run in parallel with the example coroutine .
task = asyncio . ensure_future ( client . connect ( ) )
# Wait for hangups to either finish connecting or raise an exception .
on_connect = asyncio . Future ( )
client . on_connect . add_observer ( lambda : on_connect . set_result ( None ) )
done , _ = await asyncio . wait ( ( on_connect , task ) , return_when = asyncio . FIRST_COMPLETED )
await asyncio . gather ( * done )
# Run the example coroutine . Afterwards , disconnect hangups gracefully and
# yield the hangups task to handle any exceptions .
try :
    await example_coroutine ( client , args )
except asyncio . CancelledError :
    pass
finally :
    await client . disconnect ( )
    await task


signature:
def print_table ( col_tuple , row_tuples ) :
    """Print column headers and rows as a reStructuredText table .
 Args :
 col _ tuple : Tuple of column name strings .
 row _ tuples : List of tuples containing row data ."""

implementation:
col_widths = [ max ( len ( str ( row [ col ] ) ) for row in [ col_tuple ] + row_tuples ) for col in range ( len ( col_tuple ) ) ]
format_str = ' ' . join ( '{{:<{}}}' . format ( col_width ) for col_width in col_widths )
header_border = ' ' . join ( '=' * col_width for col_width in col_widths )
print ( header_border )
print ( format_str . format ( * col_tuple ) )
print ( header_border )
for row_tuple in row_tuples :
    print ( format_str . format ( * row_tuple ) )
print ( header_border )
print ( )


signature:
def generate_enum_doc ( enum_descriptor , locations , path , name_prefix = '' ) :
    """Generate doc for an enum .
 Args :
 enum _ descriptor : descriptor _ pb2 . EnumDescriptorProto instance for enum
 to generate docs for .
 locations : Dictionary of location paths tuples to
 descriptor _ pb2 . SourceCodeInfo . Location instances .
 path : Path tuple to the enum definition .
 name _ prefix : Optional prefix for this enum ' s name ."""

implementation:
print ( make_subsection ( name_prefix + enum_descriptor . name ) )
location = locations [ path ]
if location . HasField ( 'leading_comments' ) :
    print ( textwrap . dedent ( location . leading_comments ) )
row_tuples = [ ]
for value_index , value in enumerate ( enum_descriptor . value ) :
    field_location = locations [ path + ( 2 , value_index ) ]
    row_tuples . append ( ( make_code ( value . name ) , value . number , textwrap . fill ( get_comment_from_location ( field_location ) , INFINITY ) , ) )
print_table ( ( 'Name' , 'Number' , 'Description' ) , row_tuples )


signature:
def generate_message_doc ( message_descriptor , locations , path , name_prefix = '' ) :
    """Generate docs for message and nested messages and enums .
 Args :
 message _ descriptor : descriptor _ pb2 . DescriptorProto instance for message
 to generate docs for .
 locations : Dictionary of location paths tuples to
 descriptor _ pb2 . SourceCodeInfo . Location instances .
 path : Path tuple to the message definition .
 name _ prefix : Optional prefix for this message ' s name ."""

implementation:
# message _ type is 4
prefixed_name = name_prefix + message_descriptor . name
print ( make_subsection ( prefixed_name ) )
location = locations [ path ]
if location . HasField ( 'leading_comments' ) :
    print ( textwrap . dedent ( location . leading_comments ) )
row_tuples = [ ]
for field_index , field in enumerate ( message_descriptor . field ) :
    field_location = locations [ path + ( 2 , field_index ) ]
    if field . type not in [ 11 , 14 ] :
        type_str = TYPE_TO_STR [ field . type ]
    else :
        type_str = make_link ( field . type_name . lstrip ( '.' ) )
    row_tuples . append ( ( make_code ( field . name ) , field . number , type_str , LABEL_TO_STR [ field . label ] , textwrap . fill ( get_comment_from_location ( field_location ) , INFINITY ) , ) )
print_table ( ( 'Field' , 'Number' , 'Type' , 'Label' , 'Description' ) , row_tuples )
# Generate nested messages
nested_types = enumerate ( message_descriptor . nested_type )
for index , nested_message_desc in nested_types :
    generate_message_doc ( nested_message_desc , locations , path + ( 3 , index ) , name_prefix = prefixed_name + '.' )
# Generate nested enums
for index , nested_enum_desc in enumerate ( message_descriptor . enum_type ) :
    generate_enum_doc ( nested_enum_desc , locations , path + ( 4 , index ) , name_prefix = prefixed_name + '.' )


signature:
def compile_protofile ( proto_file_path ) :
    """Compile proto file to descriptor set .
 Args :
 proto _ file _ path : Path to proto file to compile .
 Returns :
 Path to file containing compiled descriptor set .
 Raises :
 SystemExit if the compilation fails ."""

implementation:
out_file = tempfile . mkstemp ( ) [ 1 ]
try :
    subprocess . check_output ( [ 'protoc' , '--include_source_info' , '--descriptor_set_out' , out_file , proto_file_path ] )
except subprocess . CalledProcessError as e :
    sys . exit ( 'protoc returned status {}' . format ( e . returncode ) )
return out_file


signature:
def main ( ) :
    """Parse arguments and print generated documentation to stdout ."""

implementation:
parser = argparse . ArgumentParser ( )
parser . add_argument ( 'protofilepath' )
args = parser . parse_args ( )
out_file = compile_protofile ( args . protofilepath )
with open ( out_file , 'rb' ) as proto_file : # pylint : disable = no - member
    file_descriptor_set = descriptor_pb2 . FileDescriptorSet . FromString ( proto_file . read ( ) )
    # pylint : enable = no - member
for file_descriptor in file_descriptor_set . file : # Build dict of location tuples
    locations = { }
    for location in file_descriptor . source_code_info . location :
        locations [ tuple ( location . path ) ] = location
    # Add comment to top
    print ( make_comment ( 'This file was automatically generated from {} and ' 'should not be edited directly.' . format ( args . protofilepath ) ) )
    # Generate documentation
    for index , message_desc in enumerate ( file_descriptor . message_type ) :
        generate_message_doc ( message_desc , locations , ( 4 , index ) )
    for index , enum_desc in enumerate ( file_descriptor . enum_type ) :
        generate_enum_doc ( enum_desc , locations , ( 5 , index ) )


signature:
def dir_maker ( path ) :
    """Create a directory if it does not exist ."""

implementation:
directory = os . path . dirname ( path )
if directory != '' and not os . path . isdir ( directory ) :
    try :
        os . makedirs ( directory )
    except OSError as e :
        sys . exit ( 'Failed to create directory: {}' . format ( e ) )


signature:
def main ( ) :
    """Main entry point ."""

implementation:
# Build default paths for files .
dirs = appdirs . AppDirs ( 'hangups' , 'hangups' )
default_log_path = os . path . join ( dirs . user_log_dir , 'hangups.log' )
default_token_path = os . path . join ( dirs . user_cache_dir , 'refresh_token.txt' )
default_config_path = 'hangups.conf'
user_config_path = os . path . join ( dirs . user_config_dir , 'hangups.conf' )
# Create a default empty config file if does not exist .
dir_maker ( user_config_path )
if not os . path . isfile ( user_config_path ) :
    with open ( user_config_path , 'a' ) as cfg :
        cfg . write ( "" )
parser = configargparse . ArgumentParser ( prog = 'hangups' , default_config_files = [ default_config_path , user_config_path ] , formatter_class = configargparse . ArgumentDefaultsHelpFormatter , add_help = False , # Disable help so we can add it to the correct group .
)
general_group = parser . add_argument_group ( 'General' )
general_group . add ( '-h' , '--help' , action = 'help' , help = 'show this help message and exit' )
general_group . add ( '--token-path' , default = default_token_path , help = 'path used to store OAuth refresh token' )
general_group . add ( '--date-format' , default = '< %y-%m-%d >' , help = 'date format string' )
general_group . add ( '--time-format' , default = '(%I:%M:%S %p)' , help = 'time format string' )
general_group . add ( '-c' , '--config' , help = 'configuration file path' , is_config_file = True , default = user_config_path )
general_group . add ( '-v' , '--version' , action = 'version' , version = 'hangups {}' . format ( hangups . __version__ ) )
general_group . add ( '-d' , '--debug' , action = 'store_true' , help = 'log detailed debugging messages' )
general_group . add ( '--manual-login' , action = 'store_true' , help = 'enable manual login method' )
general_group . add ( '--log' , default = default_log_path , help = 'log file path' )
key_group = parser . add_argument_group ( 'Keybindings' )
key_group . add ( '--key-next-tab' , default = 'ctrl d' , help = 'keybinding for next tab' )
key_group . add ( '--key-prev-tab' , default = 'ctrl u' , help = 'keybinding for previous tab' )
key_group . add ( '--key-close-tab' , default = 'ctrl w' , help = 'keybinding for close tab' )
key_group . add ( '--key-quit' , default = 'ctrl e' , help = 'keybinding for quitting' )
key_group . add ( '--key-menu' , default = 'ctrl n' , help = 'keybinding for context menu' )
key_group . add ( '--key-up' , default = 'k' , help = 'keybinding for alternate up key' )
key_group . add ( '--key-down' , default = 'j' , help = 'keybinding for alternate down key' )
key_group . add ( '--key-page-up' , default = 'ctrl b' , help = 'keybinding for alternate page up' )
key_group . add ( '--key-page-down' , default = 'ctrl f' , help = 'keybinding for alternate page down' )
notification_group = parser . add_argument_group ( 'Notifications' )
# deprecated in favor of - - notification - type = none :
notification_group . add ( '-n' , '--disable-notifications' , action = 'store_true' , help = configargparse . SUPPRESS )
notification_group . add ( '-D' , '--discreet-notifications' , action = 'store_true' , help = 'hide message details in notifications' )
notification_group . add ( '--notification-type' , choices = sorted ( NOTIFIER_TYPES . keys ( ) ) , default = 'default' , help = 'type of notifications to create' )
# add color scheme options
col_group = parser . add_argument_group ( 'Colors' )
col_group . add ( '--col-scheme' , choices = COL_SCHEMES . keys ( ) , default = 'default' , help = 'colour scheme to use' )
col_group . add ( '--col-palette-colors' , choices = ( '16' , '88' , '256' ) , default = 16 , help = 'Amount of available colors' )
for name in COL_SCHEME_NAMES :
    col_group . add ( '--col-' + name . replace ( '_' , '-' ) + '-fg' , help = name + ' foreground color' )
    col_group . add ( '--col-' + name . replace ( '_' , '-' ) + '-bg' , help = name + ' background color' )
args = parser . parse_args ( )
# Create all necessary directories .
for path in [ args . log , args . token_path ] :
    dir_maker ( path )
logging . basicConfig ( filename = args . log , level = logging . DEBUG if args . debug else logging . WARNING , format = LOG_FORMAT )
# urwid makes asyncio ' s debugging logs VERY noisy , so adjust the log level :
logging . getLogger ( 'asyncio' ) . setLevel ( logging . WARNING )
datetimefmt = { 'date' : args . date_format , 'time' : args . time_format }
# setup color scheme
palette_colors = int ( args . col_palette_colors )
col_scheme = COL_SCHEMES [ args . col_scheme ]
for name in COL_SCHEME_NAMES :
    col_scheme = add_color_to_scheme ( col_scheme , name , getattr ( args , 'col_' + name + '_fg' ) , getattr ( args , 'col_' + name + '_bg' ) , palette_colors )
keybindings = { 'next_tab' : args . key_next_tab , 'prev_tab' : args . key_prev_tab , 'close_tab' : args . key_close_tab , 'quit' : args . key_quit , 'menu' : args . key_menu , 'up' : args . key_up , 'down' : args . key_down , 'page_up' : args . key_page_up , 'page_down' : args . key_page_down , }
notifier_ = get_notifier ( args . notification_type , args . disable_notifications )
try :
    ChatUI ( args . token_path , keybindings , col_scheme , palette_colors , datetimefmt , notifier_ , args . discreet_notifications , args . manual_login )
except KeyboardInterrupt :
    sys . exit ( 'Caught KeyboardInterrupt, exiting abnormally' )


signature:
def _exception_handler ( self , _loop , context ) :
    """Handle exceptions from the asyncio loop ."""

implementation:
# Start a graceful shutdown .
self . _coroutine_queue . put ( self . _client . disconnect ( ) )
# Store the exception to be re - raised later . If the context doesn ' t
# contain an exception , create one containing the error message .
default_exception = Exception ( context . get ( 'message' ) )
self . _exception = context . get ( 'exception' , default_exception )


signature:
def _input_filter ( self , keys , _ ) :
    """Handle global keybindings ."""

implementation:
if keys == [ self . _keys [ 'menu' ] ] :
    if self . _urwid_loop . widget == self . _tabbed_window :
        self . _show_menu ( )
    else :
        self . _hide_menu ( )
elif keys == [ self . _keys [ 'quit' ] ] :
    self . _coroutine_queue . put ( self . _client . disconnect ( ) )
else :
    return keys


signature:
def _show_menu ( self ) :
    """Show the overlay menu ."""

implementation:
# If the current widget in the TabbedWindowWidget has a menu ,
# overlay it on the TabbedWindowWidget .
current_widget = self . _tabbed_window . get_current_widget ( )
if hasattr ( current_widget , 'get_menu_widget' ) :
    menu_widget = current_widget . get_menu_widget ( self . _hide_menu )
    overlay = urwid . Overlay ( menu_widget , self . _tabbed_window , align = 'center' , width = ( 'relative' , 80 ) , valign = 'middle' , height = ( 'relative' , 80 ) )
    self . _urwid_loop . widget = overlay


signature:
def get_conv_widget ( self , conv_id ) :
    """Return an existing or new ConversationWidget ."""

implementation:
if conv_id not in self . _conv_widgets :
    set_title_cb = ( lambda widget , title : self . _tabbed_window . set_tab ( widget , title = title ) )
    widget = ConversationWidget ( self . _client , self . _coroutine_queue , self . _conv_list . get ( conv_id ) , set_title_cb , self . _keys , self . _datetimefmt )
    self . _conv_widgets [ conv_id ] = widget
return self . _conv_widgets [ conv_id ]


signature:
def add_conversation_tab ( self , conv_id , switch = False ) :
    """Add conversation tab if not present , and optionally switch to it ."""

implementation:
conv_widget = self . get_conv_widget ( conv_id )
self . _tabbed_window . set_tab ( conv_widget , switch = switch , title = conv_widget . title )


signature:
async def _on_connect ( self ) :
    """Handle connecting for the first time ."""

implementation:
self . _user_list , self . _conv_list = ( await hangups . build_user_conversation_list ( self . _client ) )
self . _conv_list . on_event . add_observer ( self . _on_event )
# show the conversation menu
conv_picker = ConversationPickerWidget ( self . _conv_list , self . on_select_conversation , self . _keys )
self . _tabbed_window = TabbedWindowWidget ( self . _keys )
self . _tabbed_window . set_tab ( conv_picker , switch = True , title = 'Conversations' )
self . _urwid_loop . widget = self . _tabbed_window


signature:
def _on_event ( self , conv_event ) :
    """Open conversation tab for new messages & pass events to notifier ."""

implementation:
conv = self . _conv_list . get ( conv_event . conversation_id )
user = conv . get_user ( conv_event . user_id )
show_notification = all ( ( isinstance ( conv_event , hangups . ChatMessageEvent ) , not user . is_self , not conv . is_quiet , ) )
if show_notification :
    self . add_conversation_tab ( conv_event . conversation_id )
    if self . _discreet_notifications :
        notification = DISCREET_NOTIFICATION
    else :
        notification = notifier . Notification ( user . full_name , get_conv_name ( conv ) , conv_event . text )
    self . _notifier . send ( notification )


signature:
def put ( self , coro ) :
    """Put a coroutine in the queue to be executed ."""

implementation:
# Avoid logging when a coroutine is queued or executed to avoid log
# spam from coroutines that are started on every keypress .
assert asyncio . iscoroutine ( coro )
self . _queue . put_nowait ( coro )


signature:
async def consume ( self ) :
    """Consume coroutines from the queue by executing them ."""

implementation:
while True :
    coro = await self . _queue . get ( )
    assert asyncio . iscoroutine ( coro )
    await coro


signature:
def _rename ( self , name , callback ) :
    """Rename conversation and call callback ."""

implementation:
self . _coroutine_queue . put ( self . _conversation . rename ( name ) )
callback ( )


signature:
def _on_event ( self , _ ) :
    """Re - order the conversations when an event occurs ."""

implementation:
# TODO : handle adding new conversations
self . sort ( key = lambda conv_button : conv_button . last_modified , reverse = True )


signature:
def show_message ( self , message_str ) :
    """Show a temporary message ."""

implementation:
if self . _message_handle is not None :
    self . _message_handle . cancel ( )
self . _message_handle = asyncio . get_event_loop ( ) . call_later ( self . _MESSAGE_DELAY_SECS , self . _clear_message )
self . _message = message_str
self . _update ( )


signature:
def _on_event ( self , conv_event ) :
    """Make users stop typing when they send a message ."""

implementation:
if isinstance ( conv_event , hangups . ChatMessageEvent ) :
    self . _typing_statuses [ conv_event . user_id ] = ( hangups . TYPING_TYPE_STOPPED )
    self . _update ( )


signature:
def _on_typing ( self , typing_message ) :
    """Handle typing updates ."""

implementation:
self . _typing_statuses [ typing_message . user_id ] = typing_message . status
self . _update ( )


signature:
def _update ( self ) :
    """Update status text ."""

implementation:
typing_users = [ self . _conversation . get_user ( user_id ) for user_id , status in self . _typing_statuses . items ( ) if status == hangups . TYPING_TYPE_STARTED ]
displayed_names = [ user . first_name for user in typing_users if not user . is_self ]
if displayed_names :
    typing_message = '{} {} typing...' . format ( ', ' . join ( sorted ( displayed_names ) ) , 'is' if len ( displayed_names ) == 1 else 'are' )
else :
    typing_message = ''
if not self . _is_connected :
    self . _widget . set_text ( "RECONNECTING..." )
elif self . _message is not None :
    self . _widget . set_text ( self . _message )
else :
    self . _widget . set_text ( typing_message )


signature:
def _get_date_str ( timestamp , datetimefmt , show_date = False ) :
    """Convert UTC datetime into user interface string ."""

implementation:
fmt = ''
if show_date :
    fmt += '\n' + datetimefmt . get ( 'date' , '' ) + '\n'
fmt += datetimefmt . get ( 'time' , '' )
return timestamp . astimezone ( tz = None ) . strftime ( fmt )


signature:
def from_conversation_event ( conversation , conv_event , prev_conv_event , datetimefmt , watermark_users = None ) :
    """Return MessageWidget representing a ConversationEvent .
 Returns None if the ConversationEvent does not have a widget
 representation ."""

implementation:
user = conversation . get_user ( conv_event . user_id )
# Check whether the previous event occurred on the same day as this
# event .
if prev_conv_event is not None :
    is_new_day = ( conv_event . timestamp . astimezone ( tz = None ) . date ( ) != prev_conv_event . timestamp . astimezone ( tz = None ) . date ( ) )
else :
    is_new_day = False
if isinstance ( conv_event , hangups . ChatMessageEvent ) :
    return MessageWidget ( conv_event . timestamp , conv_event . text , datetimefmt , user , show_date = is_new_day , watermark_users = watermark_users )
elif isinstance ( conv_event , hangups . RenameEvent ) :
    if conv_event . new_name == '' :
        text = ( '{} cleared the conversation name' . format ( user . first_name ) )
    else :
        text = ( '{} renamed the conversation to {}' . format ( user . first_name , conv_event . new_name ) )
    return MessageWidget ( conv_event . timestamp , text , datetimefmt , show_date = is_new_day , watermark_users = watermark_users )
elif isinstance ( conv_event , hangups . MembershipChangeEvent ) :
    event_users = [ conversation . get_user ( user_id ) for user_id in conv_event . participant_ids ]
    names = ', ' . join ( [ user . full_name for user in event_users ] )
    if conv_event . type_ == hangups . MEMBERSHIP_CHANGE_TYPE_JOIN :
        text = ( '{} added {} to the conversation' . format ( user . first_name , names ) )
    else : # LEAVE
        text = ( '{} left the conversation' . format ( names ) )
    return MessageWidget ( conv_event . timestamp , text , datetimefmt , show_date = is_new_day , watermark_users = watermark_users )
elif isinstance ( conv_event , hangups . HangoutEvent ) :
    text = { hangups . HANGOUT_EVENT_TYPE_START : ( 'A Hangout call is starting.' ) , hangups . HANGOUT_EVENT_TYPE_END : ( 'A Hangout call ended.' ) , hangups . HANGOUT_EVENT_TYPE_ONGOING : ( 'A Hangout call is ongoing.' ) , } . get ( conv_event . event_type , 'Unknown Hangout call event.' )
    return MessageWidget ( conv_event . timestamp , text , datetimefmt , show_date = is_new_day , watermark_users = watermark_users )
elif isinstance ( conv_event , hangups . GroupLinkSharingModificationEvent ) :
    status_on = hangups . GROUP_LINK_SHARING_STATUS_ON
    status_text = ( 'on' if conv_event . new_status == status_on else 'off' )
    text = '{} turned {} joining by link.' . format ( user . first_name , status_text )
    return MessageWidget ( conv_event . timestamp , text , datetimefmt , show_date = is_new_day , watermark_users = watermark_users )
else : # conv _ event is a generic hangups . ConversationEvent .
    text = 'Unknown conversation event'
    return MessageWidget ( conv_event . timestamp , text , datetimefmt , show_date = is_new_day , watermark_users = watermark_users )


signature:
def _handle_event ( self , conv_event ) :
    """Handle updating and scrolling when a new event is added .
 Automatically scroll down to show the new text if the bottom is
 showing . This allows the user to scroll up to read previous messages
 while new messages are arriving ."""

implementation:
if not self . _is_scrolling :
    self . set_focus ( conv_event . id_ )
else :
    self . _modified ( )


signature:
async def _load ( self ) :
    """Load more events for this conversation ."""

implementation:
try :
    conv_events = await self . _conversation . get_events ( self . _conversation . events [ 0 ] . id_ )
except ( IndexError , hangups . NetworkError ) :
    conv_events = [ ]
if not conv_events :
    self . _first_loaded = True
if self . _focus_position == self . POSITION_LOADING and conv_events : # If the loading indicator is still focused , and we loaded more
# events , set focus on the first new event so the loaded
# indicator is replaced .
    self . set_focus ( conv_events [ - 1 ] . id_ )
else : # Otherwise , still need to invalidate in case the loading
# indicator is showing but not focused .
    self . _modified ( )
# Loading events can also update the watermarks .
self . _refresh_watermarked_events ( )
self . _is_loading = False


signature:
def _get_position ( self , position , prev = False ) :
    """Return the next / previous position or raise IndexError ."""

implementation:
if position == self . POSITION_LOADING :
    if prev :
        raise IndexError ( 'Reached last position' )
    else :
        return self . _conversation . events [ 0 ] . id_
else :
    ev = self . _conversation . next_event ( position , prev = prev )
    if ev is None :
        if prev :
            return self . POSITION_LOADING
        else :
            raise IndexError ( 'Reached first position' )
    else :
        return ev . id_


signature:
def set_focus ( self , position ) :
    """Set the focus to position or raise IndexError ."""

implementation:
self . _focus_position = position
self . _modified ( )
# If we set focus to anywhere but the last position , the user if
# scrolling up :
try :
    self . next_position ( position )
except IndexError :
    self . _is_scrolling = False
else :
    self . _is_scrolling = True


signature:
def get_menu_widget ( self , close_callback ) :
    """Return the menu widget associated with this widget ."""

implementation: return ConversationMenu ( self . _coroutine_queue , self . _conversation , close_callback , self . _keys )


signature:
def keypress ( self , size , key ) :
    """Handle marking messages as read and keeping client active ."""

implementation:
# Set the client as active .
self . _coroutine_queue . put ( self . _client . set_active ( ) )
# Mark the newest event as read .
self . _coroutine_queue . put ( self . _conversation . update_read_timestamp ( ) )
return super ( ) . keypress ( size , key )


signature:
def _set_title ( self ) :
    """Update this conversation ' s tab title ."""

implementation:
self . title = get_conv_name ( self . _conversation , show_unread = True , truncate = True )
self . _set_title_cb ( self , self . title )


signature:
def _on_return ( self , text ) :
    """Called when the user presses return on the send message widget ."""

implementation:
# Ignore if the user hasn ' t typed a message .
if not text :
    return
elif text . startswith ( '/image' ) and len ( text . split ( ' ' ) ) == 2 : # Temporary UI for testing image uploads
    filename = text . split ( ' ' ) [ 1 ]
    image_file = open ( filename , 'rb' )
    text = ''
else :
    image_file = None
text = replace_emoticons ( text )
segments = hangups . ChatMessageSegment . from_str ( text )
self . _coroutine_queue . put ( self . _handle_send_message ( self . _conversation . send_message ( segments , image_file = image_file ) ) )


signature:
def _update_tabs ( self ) :
    """Update tab display ."""

implementation:
text = [ ]
for num , widget in enumerate ( self . _widgets ) :
    palette = ( 'active_tab' if num == self . _tab_index else 'inactive_tab' )
    text += [ ( palette , ' {} ' . format ( self . _widget_title [ widget ] ) ) , ( 'tab_background' , ' ' ) , ]
self . _tabs . set_text ( text )
self . _frame . contents [ 'body' ] = ( self . _widgets [ self . _tab_index ] , None )


signature:
def keypress ( self , size , key ) :
    """Handle keypresses for changing tabs ."""

implementation:
key = super ( ) . keypress ( size , key )
num_tabs = len ( self . _widgets )
if key == self . _keys [ 'prev_tab' ] :
    self . _tab_index = ( self . _tab_index - 1 ) % num_tabs
    self . _update_tabs ( )
elif key == self . _keys [ 'next_tab' ] :
    self . _tab_index = ( self . _tab_index + 1 ) % num_tabs
    self . _update_tabs ( )
elif key == self . _keys [ 'close_tab' ] : # Don ' t allow closing the Conversations tab
    if self . _tab_index > 0 :
        curr_tab = self . _widgets [ self . _tab_index ]
        self . _widgets . remove ( curr_tab )
        del self . _widget_title [ curr_tab ]
        self . _tab_index -= 1
        self . _update_tabs ( )
else :
    return key


signature:
def set_tab ( self , widget , switch = False , title = None ) :
    """Add or modify a tab .
 If widget is not a tab , it will be added . If switch is True , switch to
 this tab . If title is given , set the tab ' s title ."""

implementation:
if widget not in self . _widgets :
    self . _widgets . append ( widget )
    self . _widget_title [ widget ] = ''
if switch :
    self . _tab_index = self . _widgets . index ( widget )
if title :
    self . _widget_title [ widget ] = title
self . _update_tabs ( )


signature:
def _replace_words ( replacements , string ) :
    """Replace words with corresponding values in replacements dict .
 Words must be separated by spaces or newlines ."""

implementation:
output_lines = [ ]
for line in string . split ( '\n' ) :
    output_words = [ ]
    for word in line . split ( ' ' ) :
        new_word = replacements . get ( word , word )
        output_words . append ( new_word )
    output_lines . append ( output_words )
return '\n' . join ( ' ' . join ( output_words ) for output_words in output_lines )


signature:
def get_auth ( credentials_prompt , refresh_token_cache , manual_login = False ) :
    """Authenticate with Google .
 Args :
 refresh _ token _ cache ( RefreshTokenCache ) : Cache to use so subsequent
 logins may not require credentials .
 credentials _ prompt ( CredentialsPrompt ) : Prompt to use if credentials
 are required to log in .
 manual _ login ( bool ) : If true , prompt user to log in through a browser
 and enter authorization code manually . Defaults to false .
 Returns :
 dict : Google session cookies .
 Raises :
 GoogleAuthError : If authentication with Google fails ."""

implementation:
with requests . Session ( ) as session :
    session . headers = { 'user-agent' : USER_AGENT }
    try :
        logger . info ( 'Authenticating with refresh token' )
        refresh_token = refresh_token_cache . get ( )
        if refresh_token is None :
            raise GoogleAuthError ( "Refresh token not found" )
        access_token = _auth_with_refresh_token ( session , refresh_token )
    except GoogleAuthError as e :
        logger . info ( 'Failed to authenticate using refresh token: %s' , e )
        logger . info ( 'Authenticating with credentials' )
        if manual_login :
            authorization_code = ( credentials_prompt . get_authorization_code ( ) )
        else :
            authorization_code = _get_authorization_code ( session , credentials_prompt )
        access_token , refresh_token = _auth_with_code ( session , authorization_code )
        refresh_token_cache . set ( refresh_token )
    logger . info ( 'Authentication successful' )
    return _get_session_cookies ( session , access_token )


signature:
def get_auth_stdin ( refresh_token_filename , manual_login = False ) :
    """Simple wrapper for : func : ` get _ auth ` that prompts the user using stdin .
 Args :
 refresh _ token _ filename ( str ) : Path to file where refresh token will be
 cached .
 manual _ login ( bool ) : If true , prompt user to log in through a browser
 and enter authorization code manually . Defaults to false .
 Raises :
 GoogleAuthError : If authentication with Google fails ."""

implementation:
refresh_token_cache = RefreshTokenCache ( refresh_token_filename )
return get_auth ( CredentialsPrompt ( ) , refresh_token_cache , manual_login = manual_login )


signature:
def _get_authorization_code ( session , credentials_prompt ) :
    """Get authorization code using Google account credentials .
 Because hangups can ' t use a real embedded browser , it has to use the
 Browser class to enter the user ' s credentials and retrieve the
 authorization code , which is placed in a cookie . This is the most fragile
 part of the authentication process , because a change to a login form or an
 unexpected prompt could break it .
 Raises GoogleAuthError authentication fails .
 Returns authorization code string ."""

implementation:
browser = Browser ( session , OAUTH2_LOGIN_URL )
email = credentials_prompt . get_email ( )
browser . submit_form ( FORM_SELECTOR , { EMAIL_SELECTOR : email } )
password = credentials_prompt . get_password ( )
browser . submit_form ( FORM_SELECTOR , { PASSWORD_SELECTOR : password } )
if browser . has_selector ( TOTP_CHALLENGE_SELECTOR ) :
    browser . submit_form ( TOTP_CHALLENGE_SELECTOR , { } )
elif browser . has_selector ( PHONE_CHALLENGE_SELECTOR ) :
    browser . submit_form ( PHONE_CHALLENGE_SELECTOR , { } )
if browser . has_selector ( VERIFICATION_FORM_SELECTOR ) :
    if browser . has_selector ( TOTP_CODE_SELECTOR ) :
        input_selector = TOTP_CODE_SELECTOR
    elif browser . has_selector ( PHONE_CODE_SELECTOR ) :
        input_selector = PHONE_CODE_SELECTOR
    else :
        raise GoogleAuthError ( 'Unknown verification code input' )
    verfification_code = credentials_prompt . get_verification_code ( )
    browser . submit_form ( VERIFICATION_FORM_SELECTOR , { input_selector : verfification_code } )
try :
    return browser . get_cookie ( 'oauth_code' )
except KeyError :
    raise GoogleAuthError ( 'Authorization code cookie not found' )


signature:
def _auth_with_refresh_token ( session , refresh_token ) :
    """Authenticate using OAuth refresh token .
 Raises GoogleAuthError if authentication fails .
 Returns access token string ."""

implementation:
# Make a token request .
token_request_data = { 'client_id' : OAUTH2_CLIENT_ID , 'client_secret' : OAUTH2_CLIENT_SECRET , 'grant_type' : 'refresh_token' , 'refresh_token' : refresh_token , }
res = _make_token_request ( session , token_request_data )
return res [ 'access_token' ]