napari-hypha-plugin-example.py

"""
This module is an example of a barebones QWidget plugin for napari

It implements the Widget specification.
see: https://napari.org/plugins/guides.html?#widgets

Replace code below according to your needs.
"""
import asyncio
from typing import TYPE_CHECKING

import numpy as np
from magicgui import magic_factory
from qtpy.QtWidgets import QHBoxLayout, QPushButton, QWidget

if TYPE_CHECKING:
    import napari


from imjoy_rpc.hypha import connect_to_server
import threading
from skimage import data

SERVER_URL = "https://ai.imjoy.io"


class ExampleQWidget(QWidget):
    # your QWidget.__init__ can optionally request the napari viewer instance
    # in one of two ways:
    # 1. use a parameter called `napari_viewer`, as done here
    # 2. use a type annotation of 'napari.viewer.Viewer' for any parameter
    def __init__(self, napari_viewer):
        super().__init__()
        self.viewer = napari_viewer

        btn = QPushButton("Click me!")
        btn.clicked.connect(self._on_click)

        self.setLayout(QHBoxLayout())
        self.layout().addWidget(btn)

        # Start an event loop in a separate thread
        def run_loop(loop):
            asyncio.set_event_loop(loop)
            loop.run_forever()

        self.loop = asyncio.new_event_loop()
        worker = threading.Thread(target=run_loop, args=(self.loop,))
        worker.setDaemon(True)
        worker.start()
        self.triton = None

    async def connect_to_server(self):
        server = await connect_to_server(
            {"name": "test client", "server_url": SERVER_URL}
        )
        self.triton = await server.get_service("triton-client")
    
    async def run_cellpose(self, evt, data):
        if not self.triton:
            await self.connect_to_server()

        results = await self.triton.execute(inputs=[data['image'], {'diameter': 30}], model_name='cellpose-python', decode_json=True)
        mask = results['mask'][0]
        data['mask'] = mask
        evt.set()

    def _on_click(self):
        image = np.random.randint(0,255, (3, 349, 467)).astype('float32')
        data = {"image": image}
        
        self.viewer.add_image(data['image'], name="image")

        evt = threading.Event()
        self.loop.call_soon_threadsafe(lambda: self.loop.create_task(self.run_cellpose(evt, data)))
        evt.wait()
        
        self.viewer.add_image(data['mask'], name="mask")

@magic_factory
def example_magic_widget(img_layer: "napari.layers.Image"):
    print(f"you have selected {img_layer}")


# Uses the `autogenerate: true` flag in the plugin manifest
# to indicate it should be wrapped as a magicgui to autogenerate
# a widget.
def example_function_widget(img_layer: "napari.layers.Image"):
    print(f"you have selected {img_layer}")