import logging
import os
import time
from timeit import default_timer as timer
from nanome._internal.network import Packet
from nanome._internal.process import ProcessManagerInstance
from nanome._internal.process import _Bonding, _Dssp
from nanome._internal.enums import Messages
from nanome._internal.decorators import deprecated
from nanome.api import shapes
from nanome.api.integration import Integration
from nanome.api.ui import Menu
from nanome.api.streams import Stream
from nanome.api import Room, Files
from nanome.util import Logs, config
from nanome.util.enums import StreamDirection, PluginListButtonType
try:
import asyncio
from nanome._internal.plugin_instance_async import async_update_loop
except ImportError:
asyncio = False
logger = logging.getLogger(__name__)
UPDATE_RATE = 1.0 / 60.0
MINIMUM_SLEEP = 0.001
# End session after 12 hours by default
# This should be long enough to indicate
# a runaway session that wasn't closed by NTS
default_session_timeout = 12 * 60 * 60
env_var_session_timeout = os.environ.get('SESSION_TIMEOUT')
if env_var_session_timeout and env_var_session_timeout.isdigit():
SESSION_TIMEOUT = int(env_var_session_timeout)
else:
SESSION_TIMEOUT = default_session_timeout
[docs]class PluginInstance:
"""
| Base class of any plugin.
| Constructor should never be called by the user as it is network-instantiated when a session connects.
| Start, update, and all methods starting by "on" can be overridden by user, in order to get requests results
"""
_instance = None
is_async = False
__callbacks = dict()
__futures = dict()
__complex_updated_callbacks = dict()
__selection_changed_callbacks = dict()
def __init__(self):
self.room = Room()
self.integration = Integration()
self.files = Files(self)
self.PluginListButtonType = PluginListButtonType
self.__set_first = False
self.__menu = Menu() # deprecated
# Make sure PluginInstance singleton is set.
PluginInstance._instance = self
[docs] def start(self):
"""
| Called when user "Activates" the plugin
"""
pass
[docs] def update(self):
"""
| Called when instance updates (multiple times per second)
"""
pass
[docs] def on_run(self):
"""
| Called when user presses "Run"
"""
Logs.warning('Callback on_run not defined. Ignoring')
[docs] def on_stop(self):
"""
| Called when user disconnects or plugin crashes
"""
pass
[docs] def on_advanced_settings(self):
"""
| Called when user presses "Advanced Settings"
"""
Logs.warning('Callback on_advanced_settings not defined. Ignoring')
[docs] def on_complex_added(self):
"""
| Called whenever a complex is added to the workspace.
"""
pass
[docs] def on_complex_removed(self):
"""
| Called whenever a complex is removed from the workspace.
"""
pass
[docs] def on_complex_list_changed(self):
"""
| Called whenever a complex is added or removed from the workspace.
"""
pass
[docs] def on_presenter_change(self):
"""
| Called when room's presenter changes.
"""
pass
[docs] def request_workspace(self, callback=None):
"""
| Request the entire workspace, in deep mode
:param callback: Callback to receive workspace.
:type callback: Callable[[Workspace], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.workspace_request, None, expects_response)
return self._save_callback(id, callback)
[docs] def request_complex_list(self, callback=None):
"""
| Request the list of all complexes in the workspace, in shallow mode.
:param callback: Callback to receive list of complexes.
:type callback: Callable[[List[Complex]], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.complex_list_request, None, expects_response)
return self._save_callback(id, callback)
[docs] def request_complexes(self, id_list, callback=None):
"""
| Requests a list of complexes by their indices
| Complexes returned contains the full structure (atom/bond/residue/chain/molecule)
:param id_list: List of indices
:type id_list: list of :class:`int`
:param callback: Callback to receive list of complexes.
:type callback: Callable[[List[Complex]], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.complexes_request, id_list, expects_response)
return self._save_callback(id, callback)
[docs] def update_workspace(self, workspace):
"""
| Replace the current workspace in the scene by the workspace in parameter
:param workspace: New workspace
:type workspace: :class:`~nanome.structure.Workspace`
"""
self._network.send(Messages.workspace_update, workspace, False)
[docs] def send_notification(self, type, message):
"""
| Send a notification to the user
:param type: Type of notification to send.
:type type: :class:`~nanome.util.enums.NotificationTypes`
:param message: Text to display to the user.
:type message: str
"""
# avoids unnecessary dependencies.
# needs to match the command serializer.
args = (type, message)
self._network.send(Messages.notification_send, args, False)
[docs] def update_structures_deep(self, structures, callback=None):
"""
| Update the specific molecular structures in the scene to match the structures in parameter.
| Will also update descendent structures and can be used to remove descendent structures.
:param structures: List of molecular structures to update.
:type structures: list of :class:`~nanome.structure.Base`
:param callback: Callback when update is done.
:type callback: Callable[[], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.structures_deep_update, structures, expects_response)
return self._save_callback(id, callback)
[docs] def update_structures_shallow(self, structures):
"""
| Update the specific molecular structures in the scene to match the structures in parameter
| Only updates the structure's data, will not update children or other descendents.
:param structures: List of molecular structures to update.
:type structures: list of :class:`~nanome.structure.Base`
"""
self._network.send(Messages.structures_shallow_update, structures, False)
[docs] def zoom_on_structures(self, structures, callback=None):
"""
| Repositions and resizes the workspace such that the provided structure(s) will be in the
| center of the users view.
:param structures: Molecular structure(s) to update.
:type structures: list of :class:`~nanome.structure.Base`
:param callback: Callback when zoom is done.
:type callback: Callable[[], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.structures_zoom, structures, expects_response)
return self._save_callback(id, callback)
[docs] def center_on_structures(self, structures, callback=None):
"""
| Repositions the workspace such that the provided structure(s) will be in the
| center of the world.
:param structures: Molecular structure(s) to update.
:type structures: list of :class:`~nanome.structure.Base`
:param callback: Callback when center is done.
:type callback: Callable[[], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.structures_center, structures, expects_response)
return self._save_callback(id, callback)
[docs] def add_to_workspace(self, complex_list, callback=None):
"""
| Add a list of complexes to the current workspace
:param complex_list: List of Complexes to add
:type complex_list: list of :class:`~nanome.structure.Complex`
:param callback: Callback when request is done.
:type callback: Callable[[], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.add_to_workspace, complex_list, expects_response)
return self._save_callback(id, callback)
[docs] def remove_from_workspace(self, complex_list, callback=None):
"""
| Remove a list of complexes from the current workspace
:param complex_list: List of Complexes to remove
:type complex_list: list of :class:`~nanome.structure.Complex`
:param callback: Callback when request is done.
:type callback: Callable[[], None]
"""
from nanome.api.structure import Complex
empty_complexes = []
for complex in complex_list:
empty = Complex()
empty.index = complex.index
empty_complexes.append(empty)
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.structures_deep_update, empty_complexes, expects_response)
return self._save_callback(id, callback)
[docs] def update_content(self, *content):
"""
| Update specific UI elements (button, slider, list...)
:param content: UI elements to update
:type content: :class:`~nanome.ui.UIBase`
or multiple :class:`~nanome.ui.UIBase`
or a list of :class:`~nanome.ui.UIBase`
"""
if len(content) == 1 and isinstance(content[0], list):
content = content[0]
self._network.send(Messages.content_update, content, False)
[docs] def update_node(self, *nodes):
"""
| Updates layout nodes and their children
:param nodes: Layout nodes to update
:type nodes: :class:`~nanome.ui.LayoutNode`
or multiple :class:`~nanome.ui.LayoutNode`
or a list of :class:`~nanome.ui.LayoutNode`
"""
if len(nodes) == 1 and isinstance(nodes[0], list):
nodes = nodes[0]
self._network.send(Messages.node_update, nodes, False)
[docs] def save_files(self, file_list, callback=None):
"""
| Save files on the machine running Nanome, and returns result
:param file_list: List of files to save with their content
:type file_list: list of :class:`~nanome.util.file.FileSaveData`
:param callback: Callback to receive save file results.
:type callable: Callable[[List[FileSaveData]], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.file_save, file_list, expects_response)
return self._save_callback(id, callback)
[docs] def create_writing_stream(self, indices_list, stream_type, callback=None):
"""
| Create a stream allowing the plugin to continuously update properties of many objects
:param indices_list: List of indices of all objects that should be in the stream
:type indices_list: list of :class:`int`
:param stream_type: Type of stream to create
:type stream_type: list of :class:`~nanome.streams.Stream.Type`
:param callback: Callback to receive the created stream.
:type callback: Callable[[Stream, StreamCreationError], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.stream_create, (stream_type, indices_list, StreamDirection.writing), expects_response)
return self._save_callback(id, callback)
[docs] def create_reading_stream(self, indices_list, stream_type, callback=None):
"""
| Create a stream allowing the plugin to continuously receive properties of many objects
:param indices_list: List of indices of all objects that should be in the stream
:type indices_list: list of :class:`int`
:param stream_type: Type of stream to create
:type stream_type: list of :class:`~nanome.streams.Stream.Type`
:param callback: Callback to receive the created stream.
:type callback: Callable[[Stream, StreamCreationError], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.stream_create, (stream_type, indices_list, StreamDirection.reading), expects_response)
return self._save_callback(id, callback)
[docs] def add_bonds(self, complex_list, callback=None, fast_mode=None):
"""
| Calculate bonds
| Requires openbabel to be installed
:param complex_list: List of complexes to add bonds to
:type complex_list: list of :class:`~nanome.structure.Complex`
:param callback: Callback when bonds are calculated.
:type callback: Callable[[List[Complex]], None]
"""
bonding = _Bonding(self, complex_list, callback, fast_mode)
return bonding._start()
[docs] def add_dssp(self, complex_list, callback=None):
"""
| Use DSSP to calculate secondary structures
:param complex_list: List of complexes to add ribbons to
:type complex_list: list of :class:`~nanome.structure.Complex`
:param callback: Callback when DSSP is calculated.
:type callback: Callable[[List[Complex]], None]
"""
dssp = _Dssp(self, complex_list, callback)
return dssp._start()
[docs] def add_volume(self, complex, volume, properties, complex_to_align_index=-1, callback=None):
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.add_volume, (complex, complex_to_align_index, volume, properties), expects_response)
return self._save_callback(id, callback)
[docs] def open_url(self, url, desktop_browser=False):
"""
| Opens a URL alongside the Nanome session in the default web browser.
:param url: url to open
:type url: str
:param desktop_browser: Whether to open the URL in the default web browser or in the Nanome session
:type desktop_browser: bool
"""
url = url.strip()
if '://' not in url:
url = 'http://' + url
self._network.send(Messages.open_url, (url, desktop_browser), False)
[docs] def request_presenter_info(self, callback=None):
"""
| Requests presenter account info (unique ID, name, email)
:param callback: Callback to receive the presenter account info.
:type callback: Callable[[PresenterInfo], None]
"""
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.presenter_info_request, None, expects_response)
return self._save_callback(id, callback)
[docs] def send_files_to_load(self, files_list, callback=None):
"""
| Send file(s) to Nanome to load directly using Nanome's importers.
| Can send just a list of paths, or a list of tuples containing (path, name)
:param files_list: List of files to load
:type files_list: list of or unique object of type :class:`str` or (:class:`str`, :class:`str`)
:param callback: Callback when files are loaded.
:type callback: Callable[[], None]
"""
files = []
if not isinstance(files_list, list):
files_list = [files_list]
for file in files_list:
if isinstance(file, tuple):
full_path, file_name = file
file_name += '.' + full_path.split('.')[-1]
else:
full_path = file.replace('\\', '/')
file_name = full_path.split('/')[-1]
with open(full_path, 'rb') as content_file:
data = content_file.read()
files.append((file_name, data))
expects_response = callback is not None or self.is_async
id = self._network.send(Messages.load_file, (files, True, True), expects_response)
return self._save_callback(id, callback)
[docs] def request_export(self, format, callback=None, entities=None):
"""
Request a file export using Nanome exporters
Can request either molecule or workspace export, for entities in Nanome workspace
or directly sent by the plugin (without begin uploaded to workspace)
:param format: File format to export
:type format: :class:`~nanome.util.enums.ExportFormats`
:param entities: Entities to export (complexes to send, or indices if referencing complexes in workspace, or a workspace, or nothing if exporting Nanome workspace)
:type entities: list of or unique object of type :class:`~nanome.structure.Workspace` or :class:`~nanome.structure.Complex`, or None, or list of or unique :class:`int`
:param callback: Callback when file is exported.
:type callback: Callable[[Union[str, bytes]], None]
"""
if entities is not None and not isinstance(entities, list):
entities = [entities]
id = self._network.send(Messages.export_files, (format, entities), True)
return self._save_callback(id, callback)
[docs] def apply_color_scheme(self, color_scheme, target, only_carbons):
"""
Applies a color scheme to selected atoms.
:param color_scheme: the color scheme to use on atoms
:type color_scheme: :class:`~nanome.util.enums.ColorScheme`
:param target: whether you want to color the atom, the surface, or the ribbon
:type target: :class:`~nanome.util.enums.ColorSchemeTarget`
:param only_carbons: whether you want to only color carbons, or all atoms.
:type only_carbons: bool
"""
self._network.send(Messages.apply_color_scheme, (color_scheme, target, only_carbons), False)
@property
def plugin_files_path(self):
path = os.path.expanduser(config.fetch('plugin_files_path'))
if not os.path.exists(path):
os.makedirs(path)
return path
@property
def custom_data(self):
"""
| Get custom data set with Plugin.set_custom_data
:type: tuple of objects or None if no data has been set
"""
return self._custom_data
@property
def menu(self):
if not self.__set_first:
self.__set_first = True
Logs.warning("The default menu (self.menu) is now deprecated and will be removed in a future version. Please use the ui.Menu() constructor to create the menu.")
return self.__menu
@menu.setter
def menu(self, value):
self.__set_first = True
self.__menu = value
[docs] @Logs.deprecated("create_writing_stream")
def create_stream(self, atom_indices_list, callback):
id = self._network.send(Messages.stream_create, (Stream.Type.position, atom_indices_list, StreamDirection.writing), callback is not None)
self._save_callback(id, callback)
[docs] @Logs.deprecated("create_writing_stream")
def create_atom_stream(self, atom_indices_list, stream_type, callback):
self.create_writing_stream(atom_indices_list, stream_type, callback)
def _setup(
self, session_id, plugin_network, pm_queue_in, pm_queue_out, log_pipe_conn,
original_version_table, custom_data, permissions):
self._menus = {}
self._run_text = "Run"
self._run_usable = True
self._advanced_settings_text = "Advanced Settings"
self._advanced_settings_usable = True
self._custom_data = custom_data
self._permissions = permissions
self._session_timeout = SESSION_TIMEOUT
self._network = plugin_network
self._process_manager = ProcessManagerInstance(pm_queue_in, pm_queue_out)
self._log_pipe_conn = log_pipe_conn
self._network.send_connect(Messages.connect, [Packet._compression_type(), original_version_table])
logger.debug("Plugin constructed for session {}".format(session_id))
# We assume that a scientist creating their own plugin should not have to remember
# to call super()
# _setup is called by the Plugin during the process launch. If init hasn't been properly run,
# call it here.
if not hasattr(self, 'integration'):
# Call base class init first.
PluginInstance.__init__(self)
# re-init child classes, so that their overrides take priority
self.__init__()
@classmethod
def _save_callback(cls, id, callback):
if callback is None:
import nanome
if asyncio and nanome.PluginInstance._instance.is_async:
loop = asyncio.get_event_loop()
future = loop.create_future()
cls.__futures[id] = future
return future
else:
cls.__callbacks[id] = lambda *_: None
else:
cls.__callbacks[id] = callback
def _call(self, id, *args):
callbacks = self.__callbacks
futures = self.__futures
if asyncio and self.is_async and futures.get(id):
futures[id].set_result(args[0] if len(args) == 1 else args)
del futures[id]
return
if id not in callbacks:
logger.warning('Received an unknown callback id: {}'.format(id))
return
callbacks[id](*args)
del callbacks[id]
@classmethod
def _hook_complex_updated(cls, index, callback):
cls.__complex_updated_callbacks[index] = callback
@classmethod
def _hook_selection_changed(cls, index, callback):
cls.__selection_changed_callbacks[index] = callback
@classmethod
def _on_complex_updated(cls, index, new_complex):
callbacks = cls.__complex_updated_callbacks
try:
callbacks[index](new_complex)
except KeyError:
logger.warning('Received an unknown updated complex index: {}'.format(index))
@classmethod
def _on_selection_changed(cls, index, new_complex):
callbacks = cls.__selection_changed_callbacks
try:
callbacks[index](new_complex)
except KeyError:
logger.warning('Received an unknown updated complex index: {}'.format(index))
def _on_stop(self):
try:
self.on_stop()
except:
logger.error("Error in on_stop function:", exc_info=1)
def _update_loop(self):
try:
self.start()
loop_start_time = timer()
last_update = timer()
while self._network._receive() and self._process_manager.update():
self.update()
dt = last_update - timer()
sleep_time = max(UPDATE_RATE - dt, MINIMUM_SLEEP)
time.sleep(sleep_time)
last_update = timer()
if last_update - loop_start_time > self._session_timeout:
raise TimeoutError()
except KeyboardInterrupt:
self._on_stop()
return
except TimeoutError:
logger.warning("Session timed out")
self._on_stop()
self._process_manager._close()
self._network._close()
return
except Exception as e:
logger.error("Uncaught {}: {}".format(type(e).__name__, e), exc_info=1)
# Give log a little time to reach destination before closing pipe
time.sleep(0.1)
self._on_stop()
self._process_manager._close()
self._network._close()
return
def _run(self):
if asyncio and self.is_async:
coro = async_update_loop(self, UPDATE_RATE, MINIMUM_SLEEP)
asyncio.run(coro)
else:
self._update_loop()
def _has_permission(self, permission):
from nanome.api._hashes import Hashes
return Hashes.PermissionRequestHashes[permission] in self._permissions
[docs] @deprecated("files.ls")
def request_directory(self, path, callback=None, pattern="*"):
"""
| Requests the content of a directory on the machine running Nanome
:param path: Path to request. E.g. "." means Nanome's running directory
:type path: str
:param pattern: Pattern to match. E.g. "*.txt" will match all .txt files. Default value is "*" (match everything)
:type pattern: str
"""
from nanome.util import DirectoryRequestOptions
options = DirectoryRequestOptions()
options._directory_name = path
options._pattern = pattern
id = self._network.send(Messages.directory_request, options, callback != None)
self._save_callback(id, callback)
[docs] @deprecated("files.get")
def request_files(self, file_list, callback=None):
"""
| Reads files on the machine running Nanome, and returns them
:param file_list: List of file name (with path) to read. E.g. ["a.sdf", "../b.sdf"] will read a.sdf in running directory, b.sdf in parent directory, and return them
:type file_list: list of :class:`str`
"""
id = self._network.send(Messages.file_request, file_list, callback != None)
self._save_callback(id, callback)
[docs] @deprecated()
def create_shape(self, shape_type):
from nanome.util.enums import ShapeType
if shape_type == ShapeType.Sphere:
return shapes.Sphere()
if shape_type == ShapeType.Line:
return shapes.Line()
raise ValueError('Parameter shape_type must be a value of nanome.util.enums.ShapeType')
[docs]class AsyncPluginInstance(PluginInstance):
"""
| Base class of any asynchronous plugin.
| Constructor should never be called by the user as it is network-instantiated when a session connects.
| All methods available to PluginInstance are available to AsyncPluginInstance.
| Decorating these methods with @async_callback will allow them to use the async keyword in their definition
"""
is_async = True
class _DefaultPlugin(PluginInstance):
def __init__(self):
pass