Concurrent Thread Plugin level vs device level.

Ok - I'm back. Before I get to my question, I just have to say IT'S WORKING! This thing totally works, processes multiple devices as expected, etc... Now I just have to refine the rest of the parts up.

There is one thing I feel I don't understand though. The plugin is working, but I'm not comfortable that I can't simulate ONE expected behavior...

(Again, writing this at work, so using my simple markup version of this...)

This is my class:

Code: Select all

class XYZ(object):
     def __init__(self, dev, logger, sleep, stopThread):
         self.dev = dev
         self.logger = logger
         self.sleep = sleep
         self.stopThread = stopThread

     def startSocket(self):
          #Basically some stuff here that gets relevant socket data from the device's config, and starts it up.

     def stopSocket(self):
          #pretty much just socket.close()

     def runDevConcurrentThread(self):

          try:
               while True:
                    self.sleep(sleeptime)
                    try:
                         # This try inside of the while loop is because if there's no data available at the socket, I still want to do stuff.
                         #read data from socket here
                         #parse and process
                    except:
                         # Do my other non-socket reading stuff here
          except self.stopThread:
               self.logger.debug("exception hit, process exiting")
               pass

     def stopDevConcurrentThread(self):
          self.stopThread = True


...and the relevant parts of my code outside of the class that refer to it.

Code: Select all

def deviceStartComm(self, dev):
     #create an instance of the class with specs for this device
     mydevice[dev.id] = XYZ(dev, self.logger, self.sleep, self.stopThread)
     #start the socket, then start the thread
     self.mydevice[dev.id].startSocket()
     Thread(target=self.mydevice[dev.id].runDevConcurrentThread()).run()

def deviceStopComm(self,dev):
     mydevice[dev.id].stopDevConcurrentThread()
     mydevice[dev.id].stopSocket()


Now at this point, everything just works great. Plugin starts and stops without problems. Individual devices start and stop without problems, and live in their own worlds.

BUT, when I call the deviceStopComm, while the device DOES stop, and does so gracefully, it does NOT process the code after the exception (where I've included some comment above). There's no actual code there, but I want to see that debug line to make sure it's doing what I think it's doing. I can NOT seem to access that part of the code.

I suspect it has to do with the sleep method and stopThread, which I think you guys have defined somewhat specially to stop the thread in this manner. It works... but doesn't process the exception.

I've tried removing the stipulation from the exception (so, just "except:") and it still doesn't catch. But again, the thread seems to shut down without any issue.

The stopDevConcurrentThread does seem to correctly alter the state of self.stopThread (alternately, from the main class I can just set mydevice[dev.id].stopThread = False and that alters the state correctly too, so the stopDevConcurrentThread is probably entirely unnecessary).

Am I doing something wrong with passing the sleep and stopThread bits in when I instantiate the class?

Thanks! Super grateful for the help so far - I probably couldn't have got this running without you guys!

[jay (support)]

Great job, you're definitely getting closer.

First, I don't think self.stopThread (from the plugin.py side) works like you think it does. It's just a boolean that's set which the internals of the Plugin base class use to throw StopThread exceptions. In your code, you're expecting an exception to be thrown but that never happens because you never raise the exception. So, what you really need is to set self.stopThread to False in your __init__ method (you don't need to pass anything in for that) and then the while loop in your runDevConcurrentThread() method would just run as long as self.stopThread is False:

Code: Select all

while not self.stopThread:
    # Do your stuff here
# self.stopThread is now true so do any cleanup before exiting

Second, self.sleep() in plugin.py is really only useful when used in plugin.py. In your code, you should just import time and then time.sleep(secs). Again, no need to pass it into the __init__ method.

You might want to consider just opening/closing the socket inside your run loop - then you don't need to separately call start/stop. Just start/stop the thread.

So, basically, I see this:

Code: Select all

class XYZ(object):
     def __init__(self, dev, logger):
         self.dev = dev
         self.logger = logger
         self.stopThread = False

     def startSocket(self):
          #Basically some stuff here that gets relevant socket data from the device's config, and starts it up.

     def stopSocket(self):
          #pretty much just socket.close()
          #just make sure that it's actually open or swallow any exceptions along those lines

     def runDevConcurrentThread(self):
         try:
               self.startSocket()
               while not self.stopThread:
                    self.sleep(sleeptime)
                    try:
                         # This try inside of the while loop is because if there's no data available at the socket, I still want to do stuff.
                         #read data from socket here
                         #parse and process
                    except:
                         # Do my other non-socket reading stuff here
          except:
               self.logger.debug("exception hit, process exiting")
          finally:
               # Make sure you don't leave any open network connections
               self.stopSocket()

     def stopDevConcurrentThread(self):
          self.stopThread = True

Then your plugin.py code only needs to create the object and start is up and then later stop it

I was looking to do something similar for a Plugin I have been working. The plugin controls Sylvania Osram lightify 'Groups' - basically a collection of smart led bulbs that can be controlled as a group. In my case I have several around the house, and I am experimenting with 'Dynamic Scenes', where by the lights will change colors periodically - as an example - An XMas scene to make all the bulbs change red/green/etc

I believe I need a thread for each 'Group' in order to manage/change them independently of other groups. I have tried some of the recommendations on this thread, but at present the code I am trying to run in parallel just uses the main plugin thread

Code: Select all

def deviceStartComm(self, dev):
       ....
       groupThread = LightifyGroupThread(dev, self.logger)
       self.deviceThreads.append(groupThread)
       Thread(target=groupThread.runDevConcurrentThread()).run()


Main difference in my code (AFAIK) is that I didn't reassign the 'dev' object to be the new Class for threading. Instead I tried to create a list of threads (1 per device) as part of my Plugin class as you see above. When I run the plugin with the code above, 'deviceStartComm' just appears to execute the 'runDevConcurrentThread()' method indefinitely using the Plugin classes main thread..

Any help is appreciated.

[jay (support)]

What does your device class look like? And we'll need to see the full deviceStartComm method as well.

Not really wanting to hijack this thread here, but thanks Jay.

Code: Select all

        ########################################
        def deviceStartComm(self, device):
                self.debugLog("Starting device: " + device.name + ", lightifyHubIpAddr=" + self.lightifyHubIpAddr)
                if device.id not in self.deviceList and device.deviceTypeId == 'lightifyGroupType':
                        props = device.pluginProps
                        self.debugLog("Device: id=" + str(device.id) + ",name: " + device.name + ", CircadianColorTempValues=" + props["CircadianColorTempValues"] + ", CircadianBrightnessValues=" + props["CircadianBrightnessValues"])
                        # Set SupportsColor property so Indigo knows device accepts color actions and should use color UI.
                        if props["SupportsRGB"] == True:
                            # this is a color bulb
                            props["SupportsColor"] = True
                        else:
                            # this is a tunable white bulb
                            props["SupportsColor"] = True
                            props["SupportsWhiteTemperature"] = True

                        device.replacePluginPropsOnServer(props)
                        self.update(device)
                        self.deviceList.append(device.id)

                        groupThread = LightifyGroupThread(device, self.logger)
                        self.deviceThreads.append(groupThread)
                        Thread(target=groupThread.runDevConcurrentThread()).run()


Right now the new class is inside my plugin.py, but defined at the same level as Plugin.. It is really just a stub class right now that I stole from earlier in this thread, hence I haven't plugged any real logic here until I can fix the threading.

Code: Select all

########################################
class LightifyGroupThread(object):
     def __init__(self, group, logger):
         self.group = group
         self.logger = logger
         self.stopThread = False

     def startScene(self):
        #Basically some stuff here that gets relevant socket data from the device's config, and starts it up.
        self.logger.debug("...start Scene action  group - " + self.group.name)

     def stopScene(self):
        #pretty much just socket.close()
        self.logger.debug("...stop Scene action  group - " + self.group.name)

     def runDevConcurrentThread(self):
          try:
                while not self.stopThread:
                    time.sleep(10)
                    try:
                         # This try inside of the while loop is because if there's no data available at the socket, I still want to do stuff.
                         #read data from socket here
                         #parse and process
                         self.logger.debug("...try something - group - " + self.group.name)
                    except:
                         # Do my other non-socket reading stuff here
                         self.logger.debug("...except something - group - " + self.group.name)

          except self.stopThread:
               self.logger.debug("exception hit, process exiting")
               pass

     def stopDevConcurrentThread(self):
          self.stopThread = True



Basically when this code run, the very first execution of 'deviceStartComm' seems to take over the main plugin thread.. It only attempts to run the first device, as 'deviceStartComm' isn't run for the remaining devices.

[jay (support)]

Your loop is likely because of this line:

Code: Select all

 device.replacePluginPropsOnServer(props)

That generally results in calling stopDeviceComm/startDeviceComm unless you've defined didDeviceCommPropertyChange() to do otherwise.

Also, it looks like you didn't make the changes I suggested above to your device class - in it's current form it will never stop cleanly because your catching an exception that will never get thrown.

So I tried commenting out the - device.replacePluginPropsOnServer(props) - method call. However, the code still seems to hijack the main plugin thread..

Basically in my class for a separate thread, I've tried 'self.sleep' vs. 'time.sleep' as shown below

Code: Select all

 def runDevConcurrentThread(self):
        try:
            while not self.stopThread:
               self.sleep(30)
                ....


When using 'self.sleep' I get an exception thrown indicating the class has no 'sleep' method ... (<type 'exceptions.AttributeError'>, AttributeError("'LightifyGroupThread' object has no attribute 'sleep'",), <traceback object at 0x10d34f2d8>)

Code: Select all

 def runDevConcurrentThread(self):
        try:
            while not self.stopThread:
                time.sleep(30)
                ....


When using 'time.sleep', the main Plugin.class method is used and the 'deviceStartComm' never completes.

Not sure what is happening.. Does it matter that my thread class is in the same file as the main Plugin?

Try these changes.

Code: Select all

        ########################################
        def deviceStartComm(self, device):
                self.debugLog("Starting device: " + device.name + ", lightifyHubIpAddr=" + self.lightifyHubIpAddr)
                if device.id not in self.deviceList and device.deviceTypeId == 'lightifyGroupType':
                        props = device.pluginProps
                        self.debugLog("Device: id=" + str(device.id) + ",name: " + device.name + ", CircadianColorTempValues=" + props["CircadianColorTempValues"] + ", CircadianBrightnessValues=" + props["CircadianBrightnessValues"])
                        self.deviceList.append(device.id)

                        groupThread = LightifyGroupThread(device, self.logger)
                        self.deviceThreads.append(groupThread)


Code: Select all

########################################
class LightifyGroupThread(object):
     def __init__(self, group, logger):
         self.group = group
         self.logger = logger
         self.stopThread = False
         self.thread = Thread(target=self.runDevConcurrentThread)
         self.thread.start()

     def startScene(self):
        #Basically some stuff here that gets relevant socket data from the device's config, and starts it up.
        self.logger.debug("...start Scene action  group - " + self.group.name)

     def stopScene(self):
        #pretty much just socket.close()
        self.logger.debug("...stop Scene action  group - " + self.group.name)

     def runDevConcurrentThread(self):
          try:
                while not self.stopThread:
                    time.sleep(1)
                    try:
                         # This try inside of the while loop is because if there's no data available at the socket, I still want to do stuff.
                         #read data from socket here
                         #parse and process
                         self.logger.debug("...try something - group - " + self.group.name)
                    except:
                         # Do my other non-socket reading stuff here
                         self.logger.debug("...except something - group - " + self.group.name)

     def stopDevConcurrentThread(self):
          self.stopThread = True


Great. That did the trick.. New threads seem to be happy and running while the main Plugin thread is still running fine.. I need to study up on python threading 101

Thanks again for the assistance Jay and FlyingDiver!

[jay (support)]

I've found the best way to do this sort of thing is to make your device-specific things subclasses of threading.Thread. So all of your initialization, configuration, and runtime elements are in the same class. But that's just me...

So let me know if this is following your best practice recommendation - basically the new class extends 'threading.Thread' - and it is responsible for starting itself .. BTW at work right now so not sure if the code below will run.

Code: Select all

class LightifyGroupThread(threading.Thread):
    def __init__(self, group, logger):
        self.group = group
        self.logger = logger
        self.stopThread = False
        self.start()

    def startScene(self):
        # start the scene
        self.logger.debug("...start Scene group - " + str(self.group.name))

    def stopScene(self):
        # stop the scene
        self.logger.debug("...stop Scene group - " + str(self.group.name))

    def run(self):
        try:
            while not self.stopThread:
                time.sleep(30)
                try:
                    # do the scene stuff
                    self.logger.debug("...Scene group running - " + str(self.group.name))

                except:
                    # Do my other non-socket reading stuff here
                    self.logger.debug("...Scene group exception - " + str(self.group.name))
                    self.logger.debug("...Unexpected loop error: - " + str(sys.exc_info()))
        except:
            self.logger.debug("...Scene exception hit, process exiting - " + str(self.group.name))
            self.logger.debug("...Unexpected non-loop error: - " + str(sys.exc_info()))

        finally:  # Make sure you don't leave any open network connections
            self.stopScene()

    def stopDevConcurrentThread(self):
        self.logger.debug("...stopping concurrent thread for group - " + str(self.group.name))
        self.stopThread = True


[jay (support)]

Yeah, looks about right.

Note that you don't have to have it self-start if that doesn't make sense. I guess one question I have is this: does the group represented by this class always need to run/process or does it only run/process while the scene is started? I think the question is does it make sense to have both a stopScene and a stopDevConcurrentThread (the name probably could be simplified)? If this object has no use after the stopScene is called, then those two could probably be merged. If not, then it probably makes sense to not have it self-start from the init method.

But that's just food for thought really...

yeah that is a good point Jay.. For enacting dynamic scenes i probably don't really need Threads to exist and run forever. A better design may be to create Threads on an as-needed basis for the life of the scene execution. Then destroy them once the scene is stopped. I'll give this some thought..

Thanks again for all the tips

Finished getting the stubbed out class which extends threading.Thread .. Last trick was with initialization of the superclass.. Here is the stubbed version:

Code: Select all

class LightifyGroupSceneExecutor(threading.Thread):
    def __init__(self, indigoDevice, logger, sceneName):
        super(LightifyGroupSceneExecutor, self).__init__()
        self.indigoDevice = indigoDevice
        self.logger = logger
        self.stopThread = False
        self.sceneName = sceneName
        self.start()

    def startScene(self):
        # start the scene
        self.logger.debug(
            "...start Scene " + self.sceneName + " for group - " + str(self.indigoDevice.name) + ", thread-id=" + str(
                self.name))

    def stopScene(self):
        # stop the scene
        self.logger.debug(
            "...stop Scene " + self.sceneName + " for group - " + str(self.indigoDevice.name) + ", thread-id=" + str(
                self.name))

    def run(self):
        try:
            self.startScene()
            while not self.stopThread:
                time.sleep(10)
                try:
                    # do the scene stuff
                    self.logger.debug("...Scene " + self.sceneName + " group running - " + str(
                        self.indigoDevice.name) + ", thread-id=" + str(self.name))

                except:
                    # Do my exception stuff here
                    self.logger.debug(
                        "...Scene group exception - " + str(self.indigoDevice.name) + ", thread-id=" + str(self.name))
                    self.logger.debug(
                        "...Unexpected loop error: - " + str(sys.exc_info()) + ", thread-id=" + str(self.name))

        except:
            self.logger.debug(
                "...Scene exception hit, process exiting - " + str(self.indigoDevice.name) + ", thread-id=" + str(self.name))
            self.logger.debug("...Unexpected non-loop error: - " + str(sys.exc_info()) + ", thread-id=" + str(self.name))

        finally:  # Make sure you don't leave any open network connections
            self.stopScene()

    def stopDevConcurrentThread(self):
        self.logger.debug(
            "...stopping concurrent thread for group - " + str(self.indigoDevice.name) + ", thread-id=" + str(self.name))
        self.stopThread = True


Now I am spinning of new threads at will from within my Plugin. However, I quickly found out that something else in my stack isn't quite ready for multi-threading - it is either my Lightify Hub or the open source Python class I am using to send/recv commands to/from the Hub. Keep getting 'socket timeouts' when I have a few simultaneous requests going.

Anyhow, thanks again for the assistance on this one. Much appreciated

[jay (support)]

However, I quickly found out that something else in my stack isn't quite ready for multi-threading - it is either my Lightify Hub or the open source Python class I am using to send/recv commands to/from the Hub. Keep getting 'socket timeouts' when I have a few simultaneous requests going.

If that's the problem then you can always create a singleton class that is used only for the comms so that it serializes read/writes. Kinda messy, but that's the general approach when you have a resource constraint like that. Your device class would create queues (one for sending commands one for receiving) that you would then pass into the singleton when you create an instance of the device class. Then the device class just reads/writes to the queues.