Update Python example with new API

This commit is contained in:
Grégory Soutadé 2020-11-18 16:20:28 +01:00
parent d756ef2c1e
commit 389e6a108a
3 changed files with 433 additions and 55 deletions

View File

@ -20,10 +20,17 @@ import gi
gi.require_version('Gst', '1.0') gi.require_version('Gst', '1.0')
from gi.repository import GObject, GLib, Gst from gi.repository import GObject, GLib, Gst
#
# DBUS interface
#
class GenericMonitor: class GenericMonitor:
"""
Class that manage DBUS communication with GNOME generic monitor addon.
You have to subclass it
"""
def setupMonitor(self): def setupMonitor(self):
""" Setup DBUS stuff (equivalent to constructor) """
self._activated = True self._activated = True
self._encoder = json.JSONEncoder() self._encoder = json.JSONEncoder()
self._dbus_interface = 'com.soutade.GenericMonitor' self._dbus_interface = 'com.soutade.GenericMonitor'
@ -33,108 +40,462 @@ class GenericMonitor:
self.systray_proxy = self._dbus.get_object('org.gnome.Shell', '/com/soutade/GenericMonitor') self.systray_proxy = self._dbus.get_object('org.gnome.Shell', '/com/soutade/GenericMonitor')
self._dbus.add_signal_receiver(self.onClick, 'onClick', self._dbus_interface) self.add_signal_receiver(self.onClick, 'onClick')
self._dbus.add_signal_receiver(self.onDblClick, 'onDblClick', self._dbus_interface) self.add_signal_receiver(self.onDblClick, 'onDblClick')
self._dbus.add_signal_receiver(self.onRightClick, 'onRightClick', self._dbus_interface) self.add_signal_receiver(self.onRightClick, 'onRightClick')
self._dbus.add_signal_receiver(self.onDblRightClick, 'onDblRightClick', self._dbus_interface) self.add_signal_receiver(self.onRightDblClick, 'onRightDblClick')
self.add_signal_receiver(self.onScrollUp, 'onScrollUp')
self.add_signal_receiver(self.onScrollDown, 'onScrollDown')
self.add_signal_receiver(self.onEnter, 'onEnter')
self.add_signal_receiver(self.onLeave, 'onLeave')
self._dbus.add_signal_receiver(self.onActivate, 'onActivate', self._dbus_interface) self.add_signal_receiver(self.onActivate, 'onActivate')
self._dbus.add_signal_receiver(self.onDeactivate, 'onDeactivate', self._dbus_interface) self.add_signal_receiver(self.onDeactivate, 'onDeactivate')
def runMainLoop(self): def runMainLoop(self):
""" Start infinite loop that allows to send and receive events and functions """
self._mainLoop = GLib.MainLoop() self._mainLoop = GLib.MainLoop()
self._mainLoop.run() self._mainLoop.run()
def stopMainLoop(self): def stopMainLoop(self):
""" Stop infinite main loop """
self._mainLoop.quit() self._mainLoop.quit()
# Generic Monitor functions # Generic Monitor functions
def notify(self, group): def notify(self, group):
""" Send notify() function
Parameters
----------
group : GenericMonitorGroup
group to notify
"""
if self._activated: if self._activated:
if type(group) == GenericMonitorGroup: if type(group) == GenericMonitorGroup:
group = group.getValues() group = group.getValues()
self.systray_proxy.notify(self._encoder.encode(group), dbus_interface=self._dbus_interface) self.systray_proxy.notify(self._encoder.encode(group), dbus_interface=self._dbus_interface)
def deleteItems(self, items): def deleteItems(self, items):
""" Send deleteItems() function
Parameters
----------
items : list of str (<itemName>@<groupName>)
items to delete
"""
if self._activated: if self._activated:
items = {'items':items}
self.systray_proxy.deleteItems(self._encoder.encode(items), dbus_interface=self._dbus_interface) self.systray_proxy.deleteItems(self._encoder.encode(items), dbus_interface=self._dbus_interface)
def deleteGroups(self, groups): def deleteGroups(self, groups):
""" Send deleteGroups() function
Parameters
----------
groups : list of str (<groupName>)
groups to delete
"""
if self._activated: if self._activated:
groups = {'groups':groups}
self.systray_proxy.deleteGroups(self._encoder.encode(groups), dbus_interface=self._dbus_interface) self.systray_proxy.deleteGroups(self._encoder.encode(groups), dbus_interface=self._dbus_interface)
def openPopup(self, item):
""" Send openPopup() function
Parameters
----------
item : str (<itemName>@<groupName>)
Open popup (if there is one) of item
"""
if self._activated:
item = {'item':item}
self.systray_proxy.openPopup(self._encoder.encode(item), dbus_interface=self._dbus_interface)
def closePopup(self, item):
""" Send closePopup() function
Parameters
----------
item : str (<itemName>@<groupName>)
Close popup (if there is one) of item
"""
if self._activated:
item = {'item':item}
self.systray_proxy.closePopup(self._encoder.encode(item), dbus_interface=self._dbus_interface)
def togglePopup(self, item):
""" Send togglePopup() function
Parameters
----------
item : str (<itemName>@<groupName>)
Toggle popup (if there is one) of item
"""
if self._activated:
item = {'item':item}
self.systray_proxy.togglePopup(self._encoder.encode(item), dbus_interface=self._dbus_interface)
# Generic Monitor signals # Generic Monitor signals
def onClick(self, sender): def onClick(self, sender):
""" onClick event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass pass
def onRightClick(self, sender): def onRightClick(self, sender):
""" onRightClick event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass pass
def onDblClick(self, sender): def onDblClick(self, sender):
""" onDblClick event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass pass
def onDblRightClick(self, sender): def onRightDblClick(self, sender):
""" onRightDblClick event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass
def onEnter(self, sender):
""" onEnter event (mouse enter in item)
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass
def onLeave(self, sender):
""" onLeave event (mouse leave item)
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass
def onScrollUp(self, sender):
""" onScrollUp event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass
def onScrollDown(self, sender):
""" onScrollDown event
Parameters
----------
sender : str (<itemName>@<groupName>)
Sender which event has been raised
"""
pass pass
def onActivate(self): def onActivate(self):
""" onActivate event (addon activated)
"""
self._activated = True self._activated = True
def onDeactivate(self): def onDeactivate(self):
""" onDeactivate event (addon deactivated)
"""
self._activated = False self._activated = False
# DBUS method # DBUS method
def add_signal_receiver(self, callback, signalName, interface): """ Add callback when DBUS signal is raised
Parameters
----------
callback : function
Callback raised on DBUS signal
signalName : str
Name of DBUS signal
interface : str
Name of DBUS interface
"""
def add_signal_receiver(self, callback, signalName, interface=None):
if not interface:
interface = self._dbus_interface
self._dbus.add_signal_receiver(callback, signalName, interface) self._dbus.add_signal_receiver(callback, signalName, interface)
class GenericMonitorItem: #
def __init__(self, name, text='', style='', icon='', iconStyle='', onClick='', box=''): # Item stuff
self.name = name #
class GenericMonitorGenericWidget:
""" Generic widget class, parent of all widgets
"""
def __init__(self):
self.valuesToMap = []
self.mapValues = {}
self.mapName = ''
def _toMap(self):
""" Return dictionary of class elements to send to addon
"""
self.mapValues = {}
for p in self.valuesToMap:
if self.__dict__[p]:
self.mapValues[p] = self.__dict__[p]
return {self.mapName:self.mapValues}
class GenericMonitorTextWidget(GenericMonitorGenericWidget):
""" Text widget
"""
def __init__(self, text, style=''):
"""
Parameters
----------
text : str
Text to display
style : str, optional
CSS style
"""
self.valuesToMap = ('text', 'style')
self.mapName = 'text'
self.text = text self.text = text
self.style= style self.style = style
self.icon = icon
self.iconStyle = iconStyle def setText(self, text):
self.onClick = onClick self.text = text
def setStyle(self, style):
self.style = style
class GenericMonitorIconWidget(GenericMonitorGenericWidget):
""" Icon widget
"""
def __init__(self, path, style=''):
"""
Parameters
----------
path : str
Icon path
style : str, optional
CSS style
"""
self.valuesToMap = ('path', 'style')
self.mapName = 'icon'
self.path = path
self.style = style
def setPath(self, path):
self.path = path
def setStyle(self, style):
self.style = style
class GenericMonitorPictureWidget(GenericMonitorIconWidget):
""" Picture widget
"""
def __init__(self, path, style='', width=-1, height=-1):
"""
Parameters
----------
path : str
Picture path
style : str, optional
CSS style
width : int, optional
Width of displayed picture (-1 for default width)
height : int, optional
Width of displayed picture (-1 for default width)
"""
super().__init__(path, style)
self.valuesToMap = ('path', 'style', 'width', 'height')
self.mapName = 'picture'
self.width = width
self.height = height
def setWidth(self, width):
self.width = width
def setHeight(self, height):
self.height = height
class GenericMonitorPopup(GenericMonitorGenericWidget):
""" Popup of current item
"""
def __init__(self, items):
"""
Parameters
----------
items : list of GenericMonitorTextWidget and GenericMonitorPictureWidget
List of items (text or picture)
"""
self.valuesToMap = ('items',)
self.mapName = 'popup'
self.items = items
def _toMap(self):
self.mapValues = {}
self.mapValues['items'] = []
for item in self.items:
self.mapValues['items'] += [item._toMap()]
return {self.mapName:self.mapValues}
def clear(self):
""" Clear items list
"""
self.items = []
def setItems(self, items):
self.items = items
class GenericMonitorItem:
""" Addon item that will be displayed in status bar
"""
def __init__(self, name, items=[], signals={}, popup=None, box='center'):
"""
Parameters
----------
name : str
Item name
items : list of GenericMonitorTextWidget and GenericMonitorIconWidget, optional
List of items (text or icon)
signals : dictionary, optional
Dictionary of signals and their action
"on-click" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-dblclick" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-rightclick" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-rightdblclick" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-click" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-enter" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-leave" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
"on-scroll" : ["signal"|"delete"|"open-popup"|"close-popup"|"toggle-popup"]
popup : GenericMonitorPopup, optional
Popup to be displayed
box : str, optional
Box were to put items : left, center (default), or right
"""
self.name = name
self.items = items
self.signals = signals
self.popup = popup
self.box = box self.box = box
self.group = ''
self._checkValues() self._checkValues()
def _checkValues(self): def _checkValues(self):
if self.onClick and not self.onClick in ('signal', 'delete'): if not self.name:
raise ValueError('Invalid onClick value') raise ValueError('Need a name')
if len(self.items) > 2:
raise ValueError('Maximum 2 items can be displayed')
for (name, value) in self.signals.items():
if not name in ('on-click', 'on-dblclick', 'on-rightclick', 'on-rightdblclick',
'on-enter', 'on-leave', 'on-scroll'):
raise ValueError('Invalid signal name ' + name)
if not value in ('signal', 'delete', 'open-popup', 'close-popup', 'toggle-popup'):
raise ValueError('Invalid signal value ' + value)
for item in self.items:
if not isinstance(item, GenericMonitorGenericWidget):
raise ValueError('Invalid item ' + item)
if self.popup and not isinstance(self.popup, GenericMonitorPopup):
raise ValueError('Invalid popup object')
if self.box and not self.box in ('left', 'center', 'right'): if self.box and not self.box in ('left', 'center', 'right'):
raise ValueError('Invalid box value') raise ValueError('Invalid box value')
def toMap(self): def setGroup(self, group):
myMap = {"name":self.name} """ Set current group (automatically done when added in a group)
for p in ('text', 'style', 'icon', 'box'): Parameters
----------
group : str
Group name
"""
self.group = group
def getName(self):
return self.name
def getFullName(self):
""" return full name used by addon
"""
return '%s@%s' % (self.name, self.group)
def _toMap(self):
myMap = {}
for p in ('name', 'box'):
if self.__dict__[p]: if self.__dict__[p]:
myMap[p] = self.__dict__[p] myMap[p] = self.__dict__[p]
if self.iconStyle: for item in self.items:
myMap['icon-style'] = self.iconStyle item._toMap()
if self.onClick: myMap[item.mapName] = item.mapValues
myMap['on-click'] = self.onClick if self.popup:
self.popup._toMap()
myMap['popup'] = self.popup.mapValues
for (name, value) in self.signals.items():
myMap[name] = value
return [myMap] return [myMap]
class GenericMonitorGroup: class GenericMonitorGroup:
""" Group of items
"""
def __init__(self, name, items=[]): def __init__(self, name, items=[]):
"""
Parameters
----------
name : str
Group name
items : list of GenericMonitorItem, optional
List of items
"""
self.name = name self.name = name
self.items = []
if type(items) != list: if type(items) != list:
self.items = [items] self.addItem(items)
else: else:
self.items = items self.addItems(items)
def addItem(self, item): def addItem(self, item):
""" Add item into the groupw
Parameters
----------
item : GenericMonitorItem
Item to add
"""
item.setGroup(self.name)
self.items.append(item) self.items.append(item)
def addItems(self, items): def addItems(self, items):
""" Add items into the group
Parameters
----------
items : list of GenericMonitorItem
Items to add
"""
for item in items: for item in items:
self.addItem(item) self.addItem(item)
def clear(self):
""" Clear items list
"""
for item in items:
item.setGroup('')
self.items = []
def getValues(self): def getValues(self):
""" Returns group and its items in addon format
"""
res = {'group': self.name, 'items':[]} res = {'group': self.name, 'items':[]}
for item in self.items: for item in self.items:
res['items'] += item.toMap() res['items'] += item._toMap()
return res return res
def __str__(self): def __str__(self):

View File

@ -27,7 +27,7 @@ import getpass
from threading import Thread from threading import Thread
from signal import signal, SIGINT from signal import signal, SIGINT
import sys import sys
from genericmonitor import GenericMonitor, GenericMonitorGroup, GenericMonitorItem from genericmonitor import *
PURPLE_CONV_UPDATE_UNSEEN = 4 PURPLE_CONV_UPDATE_UNSEEN = 4
PURPLE_MESSAGE_SEND = 0 PURPLE_MESSAGE_SEND = 0
@ -66,27 +66,28 @@ class EventThread(Thread,GenericMonitor):
self.stopMainLoop() self.stopMainLoop()
def _getMail(self): def _getMail(self):
mailItem = GenericMonitorItem('mail')
mailGroup = GenericMonitorGroup('Mail', [mailItem])
address = "https://mail.google.com/mail/feed/atom" address = "https://mail.google.com/mail/feed/atom"
auth = HTTPBasicAuth(self.MAIL_ADDRESS, self._mail_password) auth = HTTPBasicAuth(self.MAIL_ADDRESS, self._mail_password)
req = requests.get(address, auth=auth) req = requests.get(address, auth=auth)
text = ''
style = ''
if req.status_code == requests.codes.ok: if req.status_code == requests.codes.ok:
dom = xml.dom.minidom.parseString(req.text) dom = xml.dom.minidom.parseString(req.text)
try: try:
nb_messages = int(dom.getElementsByTagName('fullcount')[0].firstChild.nodeValue) nb_messages = int(dom.getElementsByTagName('fullcount')[0].firstChild.nodeValue)
if nb_messages == 1: if nb_messages == 1:
mailItem.text = '1 msg' text = '1 msg'
elif nb_messages > 1: elif nb_messages > 1:
mailItem.text = '%d msgs' % (nb_messages) text = '%d msgs' % (nb_messages)
mailItem.style = 'color:white' style = 'color:white'
except Exception as e: except Exception as e:
mailItem.text = str(e) text = str(e)
else: else:
mailItem.text = 'Mail error %d' % (req.status_code) text = 'Mail error %d' % (req.status_code)
self.notify(mailGroup) self.mailWidget.setText(text)
self.mailWidget.setStyle(style)
self.notify(self.mailGroup)
def getEvents(self): def getEvents(self):
self._getMail() self._getMail()
@ -104,6 +105,14 @@ class EventThread(Thread,GenericMonitor):
self._mail_password = getpass.getpass('Enter password for address %s: ' % (self.MAIL_ADDRESS)) self._mail_password = getpass.getpass('Enter password for address %s: ' % (self.MAIL_ADDRESS))
self.mailWidget = GenericMonitorTextWidget('')
mailItem = GenericMonitorItem('mail', [self.mailWidget])
self.mailGroup = GenericMonitorGroup('Mail', [mailItem])
icon = GenericMonitorIconWidget('/usr/share/icons/hicolor/22x22/apps/pidgin.png', 'icon-size:22px')
pidginItem = GenericMonitorItem('pidgin', [icon])
self.pidginGroup = GenericMonitorGroup('Pidgin', pidginItem)
while not self._stopLoop: while not self._stopLoop:
self.getEvents() self.getEvents()
# Be more reactive on signal capture # Be more reactive on signal capture
@ -129,9 +138,7 @@ class EventThread(Thread,GenericMonitor):
pidginConversation.nbMessages = 1 pidginConversation.nbMessages = 1
def displayIcon(self): def displayIcon(self):
pidginItem = GenericMonitorItem('pidgin', icon='/usr/share/icons/hicolor/22x22/apps/pidgin.png', iconStyle='icon-size:22px') self.notify(self.pidginGroup)
pidginGroup = GenericMonitorGroup('Pidgin', pidginItem)
self.notify(pidginGroup)
def pidginConversationUpdated(self, conversation, _type): def pidginConversationUpdated(self, conversation, _type):
if _type != PURPLE_CONV_UPDATE_UNSEEN: if _type != PURPLE_CONV_UPDATE_UNSEEN:

View File

@ -29,7 +29,7 @@ import time
from threading import Thread from threading import Thread
from signal import signal, SIGINT from signal import signal, SIGINT
import sys import sys
from genericmonitor import GenericMonitor, GenericMonitorGroup, GenericMonitorItem from genericmonitor import *
class TimerThread(Thread,GenericMonitor): class TimerThread(Thread,GenericMonitor):
@ -38,20 +38,24 @@ class TimerThread(Thread,GenericMonitor):
self.stopMainLoop() self.stopMainLoop()
def _displayTimerValue(self): def _displayTimerValue(self):
item = GenericMonitorItem('timer', onClick='signal', box='right')
group = GenericMonitorGroup('Timer', item)
curValue = self.timers[self.curTimer] curValue = self.timers[self.curTimer]
item.text = '%02d:%02d' % (int(curValue/60)%60, curValue%60) text = '%02d:%02d' % (int(curValue/60)%60, curValue%60)
if curValue >= (60*60): if curValue >= (60*60):
item.text = '%02d:%s' % (int(curValue/(60*60)), item.text) text = '%02d:%s' % (int(curValue/(60*60)), text)
if self.curTimer == 0: if self.curTimer == 0:
if self.timerPaused and curValue:
style = 'color:black;background-color:white'
else:
style = 'color:white' style = 'color:white'
if curValue > (60*60): if curValue >= (60*60):
style += ';background-color:red' style += ';background-color:red'
else: else:
style = 'color:#215D9C' style = 'color:#215D9C'
item.style = style
self.notify(group) self.textWidget.setText(text)
self.textWidget.setStyle(style)
self.notify(self.monitorGroup)
def run(self): def run(self):
self.setupMonitor() self.setupMonitor()
@ -59,6 +63,12 @@ class TimerThread(Thread,GenericMonitor):
self.curTimer = 0 self.curTimer = 0
self.timerPaused = False self.timerPaused = False
self._stopLoop = False self._stopLoop = False
self.textWidget = GenericMonitorTextWidget('')
signals = {'on-click':'signal'}
self.monitorItem = GenericMonitorItem('timer', [self.textWidget], signals, box='right')
self.monitorGroup = GenericMonitorGroup('Timer', self.monitorItem)
while not self._stopLoop: while not self._stopLoop:
time.sleep(1) time.sleep(1)
if not self.timerPaused: if not self.timerPaused:
@ -66,7 +76,7 @@ class TimerThread(Thread,GenericMonitor):
self._displayTimerValue() self._displayTimerValue()
def _forMe(self, sender): def _forMe(self, sender):
return sender == 'timer@Timer' return sender == self.monitorItem.getFullName()
def onClick(self, sender): def onClick(self, sender):
if not self._forMe(sender): return if not self._forMe(sender): return