-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathipc.py
165 lines (134 loc) · 7.21 KB
/
ipc.py
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
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/env python3
# This file was developed by Tomáš Iser & Thomas Müller <[email protected]>.
# It is published under the BSD 3-Clause License within the LICENSE file.
"""
Interfaces with tev's IPC protocol to remote control tev with Python.
"""
import socket
import struct
import numpy as np
class TevIpc:
def __init__(self, hostname = "localhost", port = 14158):
self._hostname = hostname
self._port = port
self._socket = None
def __enter__(self):
if self._socket is not None:
raise Exception("Communication already started")
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # SOCK_STREAM means a TCP socket
self._socket.__enter__()
self._socket.connect((self._hostname, self._port))
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self._socket is None:
raise Exception("Communication was not started")
self._socket.__exit__(exc_type, exc_val, exc_tb)
"""
Opens an image from a specified path from the disk of the machine tev is running on.
"""
def open_image(self, path: str, channel_selector: str = "", grab_focus = True):
if self._socket is None:
raise Exception("Communication was not started")
data_bytes = bytearray()
data_bytes.extend(struct.pack("<I", 0)) # reserved for length
data_bytes.extend(struct.pack("<b", 7)) # open image (v2)
data_bytes.extend(struct.pack("<b", grab_focus)) # grab focus
data_bytes.extend(bytes(path, "UTF-8")) # image path
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes.extend(bytes(channel_selector, "UTF-8")) # channels to load
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes[0:4] = struct.pack("<I", len(data_bytes))
self._socket.sendall(data_bytes)
"""
Reloads the image with specified path from the disk of the machine tev is running on.
"""
def reload_image(self, name: str, grab_focus = True):
if self._socket is None:
raise Exception("Communication was not started")
data_bytes = bytearray()
data_bytes.extend(struct.pack("<I", 0)) # reserved for length
data_bytes.extend(struct.pack("<b", 1)) # reload image
data_bytes.extend(struct.pack("<b", grab_focus)) # grab focus
data_bytes.extend(bytes(name, "UTF-8")) # image path
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes[0:4] = struct.pack("<I", len(data_bytes))
self._socket.sendall(data_bytes)
"""
Closes a specified image.
"""
def close_image(self, name: str):
if self._socket is None:
raise Exception("Communication was not started")
data_bytes = bytearray()
data_bytes.extend(struct.pack("<I", 0)) # reserved for length
data_bytes.extend(struct.pack("<b", 2)) # close image
data_bytes.extend(bytes(name, "UTF-8")) # image name
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes[0:4] = struct.pack("<I", len(data_bytes))
self._socket.sendall(data_bytes)
"""
Create a blank image with a specified size and a specified set of channel names.
"R", "G", "B" [, "A"] is what should be used if an image is rendered.
"""
def create_image(self, name: str, width: int, height: int, channel_names = ["R", "G", "B", "A"], grab_focus = True):
if self._socket is None:
raise Exception("Communication was not started")
data_bytes = bytearray()
data_bytes.extend(struct.pack("<I", 0)) # reserved for length
data_bytes.extend(struct.pack("<b", 4)) # create image
data_bytes.extend(struct.pack("<b", grab_focus)) # grab focus
data_bytes.extend(bytes(name, "UTF-8")) # image name
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes.extend(struct.pack("<i", width)) # width
data_bytes.extend(struct.pack("<i", height)) # height
data_bytes.extend(struct.pack("<i", len(channel_names))) # number of channels
for cname in channel_names:
data_bytes.extend(bytes(cname, "ascii")) # channel name
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes[0:4] = struct.pack("<I", len(data_bytes))
self._socket.sendall(data_bytes)
"""
Updates the pixel values of a specified image region and a specified set of channels.
The `image` parameter must be laid out in row-major format, i.e. from most to least
significant: [row][col][channel], where the channel axis is optional.
"""
def update_image(self, name: str, image, channel_names = ["R", "G", "B", "A"], x = 0, y = 0, grab_focus = False, perform_tiling = True):
if self._socket is None:
raise Exception("Communication was not started")
# Break down image into tiles of manageable size for typical TCP packets
tile_size = [128, 128] if perform_tiling else image.shape[0:2]
n_channels = 1 if len(image.shape) < 3 else image.shape[2]
channel_offsets = [i for i in range(n_channels)]
channel_strides = [n_channels for _ in range(n_channels)]
if len(channel_names) < n_channels:
raise Exception("Not enough channel names provided")
for i in range(0, image.shape[0], tile_size[0]):
for j in range(0, image.shape[1], tile_size[1]):
tile = image[
i:(min(i + tile_size[0], image.shape[0])),
j:(min(j + tile_size[1], image.shape[1])),
...
]
tile_dense = np.full_like(tile, 0.0, dtype="<f")
tile_dense[...] = tile[...]
data_bytes = bytearray()
data_bytes.extend(struct.pack("<I", 0)) # reserved for length
data_bytes.extend(struct.pack("<b", 6)) # update image (v3)
data_bytes.extend(struct.pack("<b", grab_focus)) # grab focus
data_bytes.extend(bytes(name, "UTF-8")) # image name
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes.extend(struct.pack("<i", n_channels)) # number of channels
for channel_name in channel_names[0:n_channels]:
data_bytes.extend(bytes(channel_name, "UTF-8")) # channel name
data_bytes.extend(struct.pack("<b", 0)) # string terminator
data_bytes.extend(struct.pack("<i", x+j)) # x
data_bytes.extend(struct.pack("<i", y+i)) # y
data_bytes.extend(struct.pack("<i", tile_dense.shape[1])) # width
data_bytes.extend(struct.pack("<i", tile_dense.shape[0])) # height
for channel_offset in channel_offsets:
data_bytes.extend(struct.pack("<q", channel_offset)) # channel_offset
for channel_stride in channel_strides:
data_bytes.extend(struct.pack("<q", channel_stride)) # channel_stride
data_bytes.extend(tile_dense.tobytes()) # data
data_bytes[0:4] = struct.pack("<I", len(data_bytes))
self._socket.sendall(data_bytes)