import nanome
from nanome.util import Logs, Color
from nanome._internal.network import PluginNetwork
from nanome._internal.enums import Messages
try:
import asyncio
except ImportError:
asyncio = False
[docs]class Shape(object):
"""
| Base class of a shape. Used in self.create_shape(shape_type) in plugins.
:param shape_type: Enumerator representing the shape_type to create
:type shape_type: :class:`~nanome.util.enums.ShapeType`
"""
def __init__(self, shape_type):
self._index = -1
self._shape_type = shape_type
self._anchors = []
self._color = Color.Grey()
@property
def index(self):
"""
| Index of the shape
"""
return self._index
@property
def shape_type(self):
"""
| Type of shape. Currently Sphere, Line, and Label are supported.
:rtype: :class:`~nanome.util.enums.ShapeType`
"""
return self._shape_type
@property
def color(self):
"""
| Color of the shape
:param value: Color of the shape
:type value: :class:`~nanome.util.Color`
"""
return self._color
@color.setter
def color(self, value):
self._color = value
@property
def anchors(self):
"""
| Anchors of the shape
:param value: Anchors of the shape
:type value: list of :class:`~nanome.shapes.Anchor`
"""
return self._anchors
@anchors.setter
def anchors(self, value):
self._anchors = value
[docs] def upload(self, done_callback=None):
"""
| Upload the shape to the Nanome App
"""
return self._upload(done_callback)
[docs] @classmethod
def upload_multiple(cls, shapes, done_callback=None):
"""
| Upload multiple shapes to the Nanome App
"""
try:
return cls._upload_multiple(shapes, done_callback)
except TypeError:
# If upload multiple fails, upload each one at a time.
# Done as a fallback for older versions of Nanome that don't support
# upload_multiple yet
Logs.warning('upload_multiple() failed, attempting to upload one at a time.')
# Make sure fallback works for async calls
future = None
if done_callback is None and nanome.PluginInstance._instance.is_async:
loop = asyncio.get_event_loop()
future = loop.create_future()
def done_callback(args): return future.set_result(args)
results = []
def upload_callback(result):
results.append(result)
if len(results) == len(shapes):
done_callback(results)
for shape in shapes:
shape.upload(upload_callback if done_callback else None)
return future
[docs] def destroy(self, done_callback=None):
"""
| Remove the shape from the Nanome App and destroy it.
"""
return self._destroy(done_callback)
[docs] @classmethod
def destroy_multiple(cls, shapes, done_callback=None):
"""
| Remove multiple shapes from the Nanome App and destroy them.
"""
try:
return cls._destroy_multiple(shapes, done_callback)
except TypeError:
# If destroy multiple fails, upload each one at a time.
# Done as a fallback for older versions of Nanome that don't support
# destroy_multiple yet
Logs.warning('destroy_multiple() failed, attempting to destroy one at a time.')
# Make sure fallback works for async calls
future = None
if done_callback is None and nanome.PluginInstance._instance.is_async:
loop = asyncio.get_event_loop()
future = loop.create_future()
def done_callback(args): return future.set_result(args)
results = []
def destroy_callback(result):
results.append(result)
if len(results) == len(shapes):
done_callback(results)
for shape in shapes:
shape.destroy(destroy_callback if done_callback else None)
return future
def _upload(self, done_callback=None):
def set_callback(indices, results):
index = indices[0]
result = results[0]
if self._index != -1 and index != self._index:
Logs.error("SetShapeCallback received for the wrong shape")
self._index = index
if done_callback is not None:
done_callback(result)
id = PluginNetwork._instance.send(Messages.set_shape, [self], True)
result = nanome.PluginInstance._save_callback(
id, set_callback if done_callback else None)
if done_callback is None and nanome.PluginInstance._instance.is_async:
result.real_set_result = result.set_result
result.set_result = lambda args: set_callback(*args)
done_callback = lambda *args: result.real_set_result(args)
return result
@classmethod
def _upload_multiple(cls, shapes, done_callback=None):
def set_callback(indices, results):
error = False
for index, shape in zip(indices, shapes):
if shape._index != -1 and index != shape._index:
error = True
shape._index = index
if error:
Logs.error("SetShapeCallback received for the wrong shape")
if done_callback is not None:
done_callback(results)
id = PluginNetwork._instance.send(Messages.set_shape, shapes, True)
result = nanome.PluginInstance._save_callback(
id, set_callback if done_callback else None)
if done_callback is None and nanome.PluginInstance._instance.is_async:
result.real_set_result = result.set_result
result.set_result = lambda args: set_callback(*args)
done_callback = lambda *args: result.real_set_result(args)
return result
def _destroy(self, done_callback=None):
def set_callback(indices):
if done_callback is not None:
done_callback(indices)
id = PluginNetwork._instance.send(
Messages.delete_shape, [self._index], True)
result = nanome.PluginInstance._save_callback(
id, set_callback if done_callback else None)
if done_callback is None and nanome.PluginInstance._instance.is_async:
result.real_set_result = result.set_result
result.set_result = lambda args: set_callback(args)
def done_callback(args): return result.real_set_result(args)
return result
@classmethod
def _destroy_multiple(cls, shapes, done_callback=None):
def set_callback(indices):
if done_callback is not None:
done_callback(indices)
indices = [x._index for x in shapes]
id = PluginNetwork._instance.send(Messages.delete_shape, indices, True)
result = nanome.PluginInstance._save_callback(
id, set_callback if done_callback else None)
if done_callback is None and nanome.PluginInstance._instance.is_async:
result.real_set_result = result.set_result
result.set_result = lambda args: set_callback(args)
def done_callback(args): return result.real_set_result(args)
return result