epicure.appose_epyseg

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