Skip to content

Instantly share code, notes, and snippets.

@internetimagery
Created October 28, 2015 11:23
Show Gist options
  • Select an option

  • Save internetimagery/d22d7a1bb4653d124b71 to your computer and use it in GitHub Desktop.

Select an option

Save internetimagery/d22d7a1bb4653d124b71 to your computer and use it in GitHub Desktop.
Maya threaded function queue test
import time
import threading
import collections
import maya.cmds as cmds
import maya.utils as utils
class window(object):
def __init__(s):
name = "Testwindow"
s.maya = Maya()
if cmds.window(name, ex=True):
cmds.deleteUI(name)
name = cmds.window(name, t=name)
cmds.columnLayout(adj=True, h=300, w=300)
cmds.text(l="Display Text")
tx = cmds.textField()
cmds.button(l="TEST!", h=50, c=lambda x: s.maya.run(
s.test,
cmds.textField(tx, q=True, tx=True)
))
cmds.showWindow(name)
cmds.scriptJob(uid=[name, s.stop])
def stop(s):
s.maya.stop = True
def test(s, text):
print text
class Maya(object):
def __init__(s):
s.throttle = threading.Semaphore()
s.queue = collections.deque()
s.stop = False
threading.Thread(target=s.listen).start()
def listen(s):
while not s.stop:
if len(s.queue):
s.throttle.acquire()
s.fire()
time.sleep(0.01)
def fire(s):
func, args, kwargs = s.queue.popleft()
result = utils.executeInMainThreadWithResult(lambda: func(*args, **kwargs))
s.throttle.release()
if kwargs.get("callback", None): kwargs["callback"](result)
def run(s, func, *args, **kwargs):
s.queue.append((func, args, kwargs))
window()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment