-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathclient.py
More file actions
92 lines (69 loc) · 2.65 KB
/
client.py
File metadata and controls
92 lines (69 loc) · 2.65 KB
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
import Pyro5.api
import argparse
from graph_utils import bridge, topo_sort_handles, layout_nodes
import logging
import threading
td_proxy_container = [None]
rebuild_lock = threading.Lock()
@Pyro5.api.expose # Expose this class to be accessible over Pyro
class IOCallback:
def __init__(self, td_proxy):
self.td_proxy = td_proxy # Store the original proxy
@Pyro5.api.expose # Make sure to expose the method
def notify(self, args): # Changed from __call__ to a named method
print(f"Callback received: {args}")
rebuild_lock.release()
def rebuild_graph(td_proxy):
td_proxy.clear()
# Create a test network by bridging to the output handles from the I/O config.
created_nodes = bridge(td_proxy,
input_handles=[],
output_handles=[],
exclude_components=[
"io/*",
],
include_io_config=True)
# Sort and layout the created nodes
io_handles = td_proxy.get_io_handles()
all_nodes = created_nodes + io_handles["inputs"] + io_handles["outputs"]
print(f"All nodes: {all_nodes}")
sorted_handles = topo_sort_handles(td_proxy, all_nodes)
layout_nodes(td_proxy, sorted_handles)
def main():
global rebuild_flag
# Add at the top of the file, before any other imports
logging.basicConfig(
level=logging.DEBUG,
format='%(name)s - %(levelname)s - %(message)s',
force=True # This ensures we override any existing configuration
)
# Set up root logger
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=60883)
parser.add_argument("--test-network", action="store_true")
args = parser.parse_args()
uri = f"PYRO:td@localhost:{args.port}"
td_proxy = Pyro5.api.Proxy(uri)
print("Connected to TouchDesigner!")
# Create a Pyro daemon for the callback object
daemon = Pyro5.api.Daemon()
callback = IOCallback(td_proxy) # Pass the original proxy
uri = daemon.register(callback)
# Register the callback's URI instead of the function
td_proxy.register_io_callback(uri)
if args.test_network:
rebuild_graph(td_proxy)
# Start the daemon loop in a separate thread
thread = threading.Thread(target=daemon.requestLoop, daemon=True)
thread.start()
rebuild_lock.acquire()
while True:
rebuild_lock.acquire()
try:
rebuild_graph(td_proxy)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()