Streams API¶
Nanome provides writing_streams for updating Structures and Shapes in real-time
(reading_streams will be coming soon)
How to use a Stream¶
Instantiate by passing a list of structure indices, and stream type.
stream, error = await self.create_writing_stream(indices, stream_type)
- StreamTypes enumerated in
~nanome.utils.enums.StreamType
To write to a stream, create a list with all data to send to the stream.
stream.update(stream_data)
- If updating multiple objects, stream data is still a flat list of values.
[x1, y1, z1, x2, y2, z2]
- Each StreamType expects a different data format. See below for details.
StreamTypes¶
The top row of each table represents a structure index passed to the stream.
The bottom rows represent the expected data format to be sent to the stream for each index
Note that it is important to keep the stream data the correct length, otherwise it becomes misaligned, and will return the wrong results.
StreamType.complex_position_rotation¶
Sets the position for each complex in stream.
Stream updates take x,y,z values from postition Vector3, and x,y,z,w from rotation Quaternion
c1 | c2 | ||||||||||||
px | py | pz | rx | ry | rz | rw | px | py | pz | rx | ry | rz | rw |
Example Plugin¶
NAME = "Color Stream Test"
DESCRIPTION = "Cycle atom colors through the rainbow."
CATEGORY = "Test"
HAS_ADVANCED_OPTIONS = False
class StreamTest(nanome.AsyncPluginInstance):
@async_callback
async def on_run(self):
complex_indices = [comp.index for comp in await self.request_complex_list()]
complexes = await self.request_complexes(complex_indices)
# Generate list of atom indices to add to stream.
atom_indices = []
for comp in complexes:
atom_indices.extend([a.index for a in comp.atoms])
# Create a writing stream to set colors for every atom in the complexes.
stream_type = StreamType.color
stream, error = await self.create_writing_stream(atom_indices, stream_type)
# RGB values of the rainbow
roygbiv = [
(255, 0 , 0), # Red
(255, 127, 0), # Orange
(255, 255, 0), # Yellow
(0, 255, 0), # Green
(0, 0, 255), # Blue
(75, 0, 130), # Indigo
(148, 0, 211), # Violet
]
# Every half second, change the color of all the atoms
sleep_time = 0.5
color_index = 0
while True:
time.sleep(sleep_time)
stream_data = []
new_color_rgba = roygbiv[color_index]
for atom in atom_indices:
stream_data.extend(new_color_rgba)
stream.update(stream_data)
color_index = (color_index + 1) % len(roygbiv)
nanome.Plugin.setup(NAME, DESCRIPTION, CATEGORY, HAS_ADVANCED_OPTIONS, StreamTest)