# telepathy-python - Base classes defining the interfaces of the Telepathy framework # # Copyright (C) 2005, 2006 Collabora Limited # Copyright (C) 2005, 2006 Nokia Corporation # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; either # version 2.1 of the License, or (at your option) any later version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import dbus.service from telepathy.constants import (CONNECTION_HANDLE_TYPE_NONE, CHANNEL_CONTACT_SEARCH_STATE_BEFORE, CHANNEL_TEXT_MESSAGE_TYPE_NORMAL) from telepathy.errors import InvalidArgument from telepathy.interfaces import (CHANNEL_INTERFACE, CHANNEL_INTERFACE_DTMF, CHANNEL_INTERFACE_GROUP, CHANNEL_INTERFACE_HOLD, CHANNEL_INTERFACE_PASSWORD, CHANNEL_INTERFACE_TRANSFER, CHANNEL_TYPE_CONTACT_SEARCH, CHANNEL_TYPE_CONTACT_LIST, CHANNEL_TYPE_ROOM_LIST, CHANNEL_TYPE_STREAMED_MEDIA, CHANNEL_TYPE_TEXT, MEDIA_SESSION_HANDLER, MEDIA_STREAM_HANDLER) from telepathy._generated.Channel import Channel as _Channel class Channel(_Channel): def __init__(self, connection, type, handle): """ Initialise the base channel object. Parameters: connection - the parent Connection object type - interface name for the type of this channel handle - the channels handle if applicable """ self._conn = connection object_path = self._conn.get_channel_path() _Channel.__init__(self, self._conn._name, object_path) self._type = type self._handle = handle self._interfaces = set() @dbus.service.method(CHANNEL_INTERFACE, in_signature='', out_signature='') def Close(self): self.Closed() self._conn.remove_channel(self) @dbus.service.method(CHANNEL_INTERFACE, in_signature='', out_signature='s') def GetChannelType(self): """ Returns the interface name for the type of this channel. """ return self._type @dbus.service.method(CHANNEL_INTERFACE, in_signature='', out_signature='uu') def GetHandle(self): """ Returns the handle type and number if this channel represents a communication with a particular contact, room or server-stored list, or zero if it is transient and defined only by its contents. """ if self._handle: return self._handle.get_type(), self._handle else: return (CONNECTION_HANDLE_TYPE_NONE, 0) @dbus.service.method(CHANNEL_INTERFACE, in_signature='', out_signature='as') def GetInterfaces(self): """ Get the optional interfaces implemented by the channel. Returns: an array of the D-Bus interface names """ return self._interfaces from telepathy._generated.Channel_Type_Contact_Search \ import ChannelTypeContactSearch as _ChannelTypeContactSearchIface class ChannelTypeContactSearch(Channel, _ChannelTypeContactSearchIface): __doc__ = _ChannelTypeContactSearchIface.__doc__ def __init__(self, connection): """ Initialise the contact search channel. """ Channel.__init__(self, connection, CHANNEL_TYPE_CONTACT_SEARCH, 0) self._search_state = CHANNEL_CONTACT_SEARCH_STATE_BEFORE @dbus.service.method(CHANNEL_TYPE_CONTACT_SEARCH, in_signature='', out_signature='u') def GetSearchState(self): """ Returns the current state of this search channel object. One of the following values: 0 - CHANNEL_CONTACT_SEARCH_STATE_BEFORE the search has not started 1 - CHANNEL_CONTACT_SEARCH_STATE_DURING the search is in progress 2 - CHANNEL_CONTACT_SEARCH_STATE_AFTER the search has been completed Returns: an integer representing the search state """ return self._search_state @dbus.service.signal(CHANNEL_TYPE_CONTACT_SEARCH, signature='u') def SearchStateChanged(self, state): """ Emitted when the search state (as returned by the GetSearchState method) changes. Parameters: state - an integer representing the new search state """ self._search_state = state from telepathy._generated.Channel_Type_Contact_List \ import ChannelTypeContactList as _ChannelTypeContactListIface class ChannelTypeContactList(Channel, _ChannelTypeContactListIface): __doc__ = _ChannelTypeContactListIface.__doc__ def __init__(self, connection, handle): """ Initialise the channel. Parameters: connection - the parent Telepathy Connection object """ Channel.__init__(self, connection, CHANNEL_TYPE_CONTACT_LIST, handle) from telepathy._generated.Channel_Type_Streamed_Media \ import ChannelTypeStreamedMedia as _ChannelTypeStreamedMediaIface class ChannelTypeStreamedMedia(Channel, _ChannelTypeStreamedMediaIface): __doc__ = _ChannelTypeStreamedMediaIface.__doc__ def __init__(self, connection, handle): """ Initialise the channel. Parameters: connection - the parent Telepathy Connection object """ Channel.__init__(self, connection, CHANNEL_TYPE_STREAMED_MEDIA, handle) from telepathy._generated.Channel_Type_Room_List \ import ChannelTypeRoomList as _ChannelTypeRoomListIface class ChannelTypeRoomList(Channel, _ChannelTypeRoomListIface): __doc__ = _ChannelTypeRoomListIface.__doc__ def __init__(self, connection): """ Initialise the channel. Parameters: connection - the parent Telepathy Connection object """ Channel.__init__(self, connection, CHANNEL_TYPE_ROOM_LIST, 0) self._listing_rooms = False self._rooms = {} @dbus.service.method(CHANNEL_TYPE_ROOM_LIST, in_signature='', out_signature='b') def GetListingRooms(self): return self._listing_rooms @dbus.service.signal(CHANNEL_TYPE_ROOM_LIST, signature='b') def ListingRooms(self, listing): self._listing_rooms = listing from telepathy._generated.Channel_Type_Text \ import ChannelTypeText as _ChannelTypeTextIface class ChannelTypeText(Channel, _ChannelTypeTextIface): __doc__ = _ChannelTypeTextIface.__doc__ def __init__(self, connection, handle): """ Initialise the channel. Parameters: connection - the parent Telepathy Connection object """ Channel.__init__(self, connection, CHANNEL_TYPE_TEXT, handle) self._pending_messages = {} self._message_types = [CHANNEL_TEXT_MESSAGE_TYPE_NORMAL] @dbus.service.method(CHANNEL_TYPE_TEXT, in_signature='', out_signature='au') def GetMessageTypes(self): """ Return an array indicating which types of message may be sent on this channel. Returns: an array of integer message types as defined above """ return self._message_types @dbus.service.method(CHANNEL_TYPE_TEXT, in_signature='au', out_signature='') def AcknowledgePendingMessages(self, ids): """ Inform the channel that you have handled messages by displaying them to the user (or equivalent), so they can be removed from the pending queue. Parameters: ids - the message to acknowledge Possible Errors: InvalidArgument (a given message ID was not found, no action taken) """ for id in ids: if id not in self._pending_messages: raise InvalidArgument("the given message ID was not found") for id in ids: del self._pending_messages[id] @dbus.service.method(CHANNEL_TYPE_TEXT, in_signature='b', out_signature='a(uuuuus)') def ListPendingMessages(self, clear): """ List the messages currently in the pending queue, and optionally remove then all. Parameters: clear - a boolean indicating whether the queue should be cleared Returns: an array of structs containing: a numeric identifier a unix timestamp indicating when the message was received an integer handle of the contact who sent the message an integer of the message type a bitwise OR of the message flags a string of the text of the message """ messages = [] for id in self._pending_messages.keys(): (timestamp, sender, type, flags, text) = self._pending_messages[id] message = (id, timestamp, sender, type, flags, text) messages.append(message) if clear: del self._pending_messages[id] messages.sort(cmp=lambda x,y:cmp(x[1], y[1])) return messages @dbus.service.signal(CHANNEL_TYPE_TEXT, signature='uuuuus') def Received(self, id, timestamp, sender, type, flags, text): self._pending_messages[id] = (timestamp, sender, type, flags, text) from telepathy._generated.Channel_Interface_Call_Merging \ import ChannelInterfaceCallMerging from telepathy._generated.Channel_Interface_Chat_State \ import ChannelInterfaceChatState from telepathy._generated.Channel_Interface_DTMF import ChannelInterfaceDTMF from telepathy._generated.Channel_Interface_Group \ import ChannelInterfaceGroup as _ChannelInterfaceGroup class ChannelInterfaceGroup(_ChannelInterfaceGroup): def __init__(self): _ChannelInterfaceGroup.__init__(self) self._group_flags = 0 self._members = set() self._local_pending = set() self._remote_pending = set() @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='u') def GetGroupFlags(self): return self._group_flags @dbus.service.signal(CHANNEL_INTERFACE_GROUP, signature='uu') def GroupFlagsChanged(self, added, removed): self._group_flags |= added self._group_flags &= ~removed @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='au') def GetMembers(self): return self._members @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='u') def GetSelfHandle(self): self_handle = self._conn.GetSelfHandle() if (self_handle in self._members or self_handle in self._local_pending or self_handle in self._remote_pending): return self_handle else: return 0 @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='au') def GetLocalPendingMembers(self): return self._local_pending @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='au') def GetRemotePendingMembers(self): return self._remote_pending @dbus.service.method(CHANNEL_INTERFACE_GROUP, in_signature='', out_signature='auauau') def GetAllMembers(self): return (self._members, self._local_pending, self._remote_pending) @dbus.service.signal(CHANNEL_INTERFACE_GROUP, signature='sauauauauuu') def MembersChanged(self, message, added, removed, local_pending, remote_pending, actor, reason): self._members.update(added) self._members.difference_update(removed) self._local_pending.update(local_pending) self._local_pending.difference_update(added) self._local_pending.difference_update(removed) self._remote_pending.update(remote_pending) self._remote_pending.difference_update(added) self._remote_pending.difference_update(removed) from telepathy._generated.Channel_Interface_Hold import ChannelInterfaceHold # ChannelInterfaceMediaSignalling is in telepathy.server.media from telepathy._generated.Channel_Interface_Password \ import ChannelInterfacePassword as _ChannelInterfacePassword class ChannelInterfacePassword(_ChannelInterfacePassword): def __init__(self): _ChannelInterfacePassword.__init__(self) self._password_flags = 0 self._password = '' @dbus.service.method(CHANNEL_INTERFACE_PASSWORD, in_signature='', out_signature='u') def GetPasswordFlags(self): return self._password_flags @dbus.service.signal(CHANNEL_INTERFACE_PASSWORD, signature='uu') def PasswordFlagsChanged(self, added, removed): self._password_flags |= added self._password_flags &= ~removed from telepathy._generated.Channel_Interface_Transfer import ChannelInterfaceTransfer