1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# -*- coding: utf-8 -*-
#
# Copyright (C) 2011 Tiger Soldier
#
# This file is part of OSD Lyrics.
#
# OSD Lyrics is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OSD Lyrics 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with OSD Lyrics. If not, see <https://www.gnu.org/licenses/>.
#
from optparse import OptionParser
import dbus
import dbus.mainloop.glib
from dbus.mainloop.glib import DBusGMainLoop
import dbus.service
from gi.repository import GLib
from .consts import DAEMON_BUS_NAME
APP_BUS_PREFIX = 'org.osdlyrics.'
dbus.mainloop.glib.threads_init()
class AlreadyRunningException(Exception):
""" Raised when a process with given bus name exists.
"""
pass
class App:
""" Basic class to create a component application for OSD Lyrics.
The application creates a mainloop, owns a DBus name, and exits when the bus
name of OSD Lyrics disappears.
Once the app is created, it will parse `sys.argv`. The accepted arguments are:
- `-w`, `--watch-daemon`: If not empty, watch the daemon with given bus name.
The default value is the bus name of OSD Lyrics
To create an component application owning the bus name ``org.osdlyrics.MyApp``,
just simply follow the code below:
app = App('MyApp')
app.run()
"""
def __init__(self, name, watch_daemon=True, singleton=True):
"""
Arguments:
- `name`: The suffix of the bus name. The full bus name is
`org.osdlyrics.` + name
- `watch_daemon`: Whether to watch daemon bus
- `singleton`: If True, raise AlreadyRunningException if the bus name
already has an owner.
"""
self._name = name
self._namewatch = None
self._watch_daemon = watch_daemon
self._loop = GLib.MainLoop()
self._conn = dbus.SessionBus(mainloop=DBusGMainLoop())
self._bus_names = []
try:
self.request_bus_name(APP_BUS_PREFIX + name,
singleton)
except dbus.NameExistsException:
raise AlreadyRunningException(
'Process with bus name %s is already running' % (
APP_BUS_PREFIX + name))
self._parse_options()
def _parse_options(self):
parser = OptionParser()
parser.add_option('-w', '--watch-daemon',
dest='watch_daemon',
action='store',
default=DAEMON_BUS_NAME,
metavar='DAEMON_BUS_NAME',
help=('A well-known bus name on DBus. Exit when the'
' name disappears. If set to empty string,'
' this player proxy will not exit.'))
options, args = parser.parse_args()
if self._watch_daemon:
self._watch_daemon_bus(options.watch_daemon)
def _watch_daemon_bus(self, name):
if name:
self._namewatch = self._conn.watch_name_owner(name,
self._daemon_name_changed)
def _daemon_name_changed(self, name):
if not name:
self._loop.quit()
if (self._namewatch is not None):
self._namewatch.cancel()
self._namewatch = None
@property
def connection(self):
"""The DBus connection of the app"""
return self._conn
@property
def loop(self):
return self._loop
def run(self):
"""
Runs the main loop
Return True if the loop is quited by the program. False if quited by Ctrl+C
"""
try:
self._loop.run()
except KeyboardInterrupt:
return False
return True
def run_on_main_thread(self, target, args=(), kwargs={}):
"""Run a callable on main thread.
This is useful for notifying a thread is finished.
"""
def timeout_func(*user_data):
target(*args, **kwargs)
return GLib.SOURCE_REMOVE
GLib.timeout_add(0, timeout_func)
def quit(self):
"""Quits the main loop"""
self._loop.quit()
def request_bus_name(self, bus_name, do_not_queue=False):
"""
Request for additional well-known name on DBus
"""
self._bus_names.append(dbus.service.BusName(bus_name,
self.connection,
do_not_queue=do_not_queue))
|
