I have a python processes that monitor and interact with a Dbus service (NetworkManager)
Currently this runs in it's own thread from the main program
import dbus.mainloop.glib
import NetworkManager
PYTHON3 = sys.version_info >= (3, 0)
if PYTHON3:
from gi.repository import GObject as gobject
from gi.repository import GLib as glib
else:
import gobject
class NetworkController(threading.Thread):
def __init__(self, run_daemon_mode=False):
# Set up DBus loop
self.loop = None
dbus.mainloop.glib.threads_init()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
if PYTHON3:
self.loop = glib.MainLoop()
else:
self.loop = gobject.MainLoop()
gobject.threads_init()
#Create the thread
super(NetworkController, self).__init__(daemon=run_daemon_mode)
def run(self):
"""
Method to run the DBus main loop (on a thread)
"""
# NetworkManager init stuff...
logger.debug("Starting Network Controller loop.")
self.loop.run()
logger.debug("Network Controller loop has exited.")
def main(argv):
args=cli(argv)
#create network controller object but don't start the thread
network_con = network_controller.NetworkController(True)
network_con.start() #
try:
while True:
I now need to add in a thread to control my bluetooth GATT service (pyBluez implementation) that also uses Dbus. How do I structure my code to let each process run in it's own thread and use Dbus.
Thanks!
I created my new class which uses Dbus and is a separate thread the same way I created my first class. Then from my main routine I called the class constructor and start each thread.
class AnotherGlibMainloop(threading.Thread):
def __init__(self, run_daemon_mode=False):
# Set up DBus loop
self.loop = None
dbus.mainloop.glib.threads_init()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
if PYTHON3:
self.loop = glib.MainLoop()
else:
self.loop = gobject.MainLoop()
gobject.threads_init()
#Create the thread
super(NetworkController, self).__init__(daemon=run_daemon_mode)
def run(self):
"""
Method to run the DBus main loop (on a thread)
"""
# NetworkManager init stuff...
logger.debug("Starting Another Glib Mainloop.")
self.loop.run()
logger.debug("Another Glib Mainloop has exited.")
def main(argv): args=cli(argv)
#create network controller object but don't start the thread
network_con = network_controller.NetworkController(True)
network_con.start() #
another_glib_mainloop = AnotherGlibMainloop(True)
another_glib_mainloop.start()
try:
while True: