epicure.appose_epyseg

 1import appose
 2import numpy as np
 3import logging
 4from importlib import resources
 5import platform
 6
 7def share_as_ndarray(img: np.ndarray) -> appose.NDArray:
 8    """Copies a NumPy array into a same-sized newly allocated block of shared memory."""
 9    shared = appose.NDArray(str(img.dtype), img.shape)
10    shared.ndarray()[:] = img
11    return shared
12        
13def go_epyseg( image, parameters, progress_bar=None, logger=None ):
14    """ Install env with napari_epyseg if necessary with appose and run epyseg on the image """
15    _logger = logger or logging.getLogger(__name__)
16    try:
17        pixi_file = resources.files("epicure.resources").joinpath("pixi.toml")
18        _logger.info("Build/Load napari_epyseg environement")
19        env = appose.pixi( pixi_file ).log_debug()
20        env = env.subscribe_output( lambda line: print("OUT:", line, end="") )
21        env = env.subscribe_error( lambda line: print("DBG:", line, end="") )
22        ## get cuda installation if Linux or Windows (tensorflow with cuda)
23        is_gpu_platform = platform.system() in ("Linux", "Windows")
24        env_name = "cuda" if is_gpu_platform else "default"
25        env = env.environment(env_name).build()
26        _logger.info(f"Environment built at: {env.base()}")
27        _logger.info( "Initiate environment to run epyseg" )
28        python = env.python().init("import numpy as np; import napari_epyseg; from napari_epyseg.call_epyseg import run_epyseg")
29
30        epyseg_script = '''
31### Script to run epyseg in another python environement, created with appose and launched as a task
32data = image.ndarray()
33#task.update(f'Start segmentation with epyseg')
34import logging
35class ApposeLogHandler(logging.Handler):
36    def emit(self, record):
37        msg = self.format(record)
38        task.update(message=msg)
39
40def setup_logger( name="epyseg_worker" ):
41    logger = logging.getLogger(name)
42    handler = ApposeLogHandler()
43    formatter = logging.Formatter('[Napari-epyseg] - %(message)s')
44    handler.setFormatter( formatter )
45    logger.addHandler(handler)
46    logger.setLevel( extras["logger_level"] )
47    return logger
48
49logger = setup_logger()
50segres = run_epyseg( data, parameters, progress_bar=extras["progress_bar"], logger=logger )
51if segres is not None:
52    ## Write the results on the shared memory object
53    image.ndarray()[:] = segres
54logger.info( "Segmentation finished" )
55'''
56        def log_listener(event):
57            """ Transfer appose task message to the main logger """
58            if event.message:
59                _logger.info(f"[task] {event.message}")
60
61        try:
62            with share_as_ndarray(image) as shared_image:
63                task = python.task(epyseg_script)
64                task.listen( log_listener )
65                task.inputs["image"] = shared_image 
66                task.inputs["parameters"] = parameters
67                ## other convenient tools to print msg, show progress..
68                task.inputs["extras"] = { "logger_name":_logger.name, "logger_level":_logger.level, "progress_bar": progress_bar }
69                _logger.info( "Start segmentation in appose service task.." )
70                task.wait_for()
71                result = shared_image.ndarray()
72                return result
73        except Exception as e:
74            raise RuntimeError("Running epyseg in separated environement failed") from e
75        finally:
76            python.close()
77    except Exception as e:
78        raise RuntimeError("Epyseg in separated environement failed") from e