Coverage for openhcs/io/fiji_stream.py: 21.7%

1""" 

2Fiji streaming backend for OpenHCS. 

3 

4Streams image data directly to Fiji/ImageJ GUI for interactive exploration. 

5Uses PyImageJ to send numpy arrays directly to running Fiji instance. 

6Requires PyImageJ, JPype, and Maven to be properly configured. 

7""" 

8 

9import logging 

10from pathlib import Path 

11from typing import Any, Union, List 

12import numpy as np 

13 

14from openhcs.io.streaming import StreamingBackend 

15from openhcs.io.backend_registry import StorageBackendMeta 

16from openhcs.constants.constants import Backend 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21class FijiStreamingBackend(StreamingBackend, metaclass=StorageBackendMeta): 

22 """ 

23 Fiji streaming backend with automatic metaclass registration. 

24 

25 Streams image data directly to Fiji/ImageJ GUI for interactive exploration. 

26 Uses PyImageJ to send numpy arrays directly to running Fiji instance. 

27 """ 

28 

29 # Backend type from enum for registration 

30 _backend_type = Backend.FIJI_STREAM.value # You'd need to add this to the Backend enum 

31 

32 def __init__(self): 

33 """Initialize the Fiji streaming backend.""" 

34 self._ij = None 

35 

36 def save(self, data: Any, file_path: Union[str, Path], **kwargs) -> None: 

37 """ 

38 Stream single image to Fiji. 

39  

40 Args: 

41 data: Image data (numpy array) 

42 file_path: Identifier for the image 

43 **kwargs: Additional metadata 

44 """ 

45 if not isinstance(data, np.ndarray): 

46 logger.warning(f"Fiji streaming requires numpy arrays, got {type(data)}") 

47 return 

48 

49 # Send image directly to Fiji GUI using PyImageJ (fail loudly) 

50 self._send_to_fiji_pyimagej(data, str(file_path), **kwargs) 

51 logger.debug(f"🔬 FIJI: Streamed {file_path} to Fiji GUI for exploration") 

52 

53 def save_batch(self, data_list: List[Any], file_paths: List[Union[str, Path]], **kwargs) -> None: 

54 """ 

55 Stream batch of images to Fiji. 

56  

57 Args: 

58 data_list: List of image data arrays 

59 file_paths: List of file path identifiers 

60 **kwargs: Additional metadata 

61 """ 

62 if len(data_list) != len(file_paths): 

63 raise ValueError("Data list and file paths must have same length") 

64 

65 for data, file_path in zip(data_list, file_paths): 

66 self.save(data, file_path, **kwargs) 

67 

68 logger.info(f"🔬 FIJI: Streamed batch of {len(data_list)} images to Fiji") 

69 

70 def _send_to_fiji_pyimagej(self, data: np.ndarray, identifier: str, **kwargs) -> None: 

71 """Send image directly to Fiji using PyImageJ.""" 

72 # Try to import PyImageJ 

73 try: 

74 import imagej 

75 except ImportError: 

76 raise ImportError("PyImageJ not available. Install with: pip install 'openhcs[viz]'") 

77 

78 # Initialize PyImageJ connection if not already done 

79 if not hasattr(self, '_ij') or self._ij is None: 

80 logger.info("🔬 FIJI: Attempting to connect via PyImageJ...") 

81 self._ij = imagej.init(mode='interactive') 

82 logger.info("🔬 FIJI: ✅ Connected to Fiji via PyImageJ") 

83 

84 # Convert numpy array to ImageJ format and display 

85 ij_image = self._ij.py.to_java(data) 

86 self._ij.ui().show(identifier, ij_image) 

87 

88 # Apply auto-contrast if requested 

89 if kwargs.get('auto_contrast', True): 

90 self._ij.op().run("enhance.contrast", ij_image) 

91 

92 logger.info(f"🔬 FIJI: ✅ Sent {identifier} to Fiji via PyImageJ") 

93 

94 

95 

96 

97 

98 

99 # Example using command line (requires Fiji installation) 

100 try: 

101 import subprocess 

102 

103 # Write macro to temporary file 

104 macro_file = self._temp_dir / "temp_macro.ijm" 

105 macro_file.write_text(macro_cmd) 

106 

107 # Execute via Fiji (adjust path as needed) 

108 fiji_path = self._get_fiji_path() 

109 if fiji_path: 

110 subprocess.run([ 

111 str(fiji_path), 

112 "--headless", 

113 "--console", 

114 "-macro", 

115 str(macro_file) 

116 ], check=False, capture_output=True) 

117 

118 except Exception as e: 

119 logger.warning(f"Failed to execute Fiji macro: {e}") 

120 

121 def _get_fiji_path(self) -> Path: 

122 """Get path to Fiji executable.""" 

123 # Try common Fiji installation paths 

124 common_paths = [ 

125 Path("/Applications/Fiji.app/Contents/MacOS/ImageJ-macosx"), # macOS 

126 Path("C:/Fiji.app/ImageJ-win64.exe"), # Windows 

127 Path("/opt/fiji/ImageJ-linux64"), # Linux 

128 Path.home() / "Fiji.app" / "ImageJ-linux64", # User installation 

129 ] 

130 

131 for path in common_paths: 

132 if path.exists(): 

133 return path 

134 

135 logger.warning("Fiji executable not found in common locations") 

136 return None 

137 

138 def cleanup(self) -> None: 

139 """Clean up temporary files and resources.""" 

140 # Clean up temporary directory 

141 if self._temp_dir and self._temp_dir.exists(): 141 ↛ anywhereline 141 didn't jump anywhere: it always raised an exception.

142 try: 

143 import shutil 

144 shutil.rmtree(self._temp_dir) 

145 logger.debug("Cleaned up Fiji temporary files") 

146 except Exception as e: 

147 logger.warning(f"Failed to cleanup Fiji temp directory: {e}") 

148 

149 self._temp_dir = None 

150 self._macro_queue.clear() 

151 

152 logger.debug("Fiji streaming backend cleaned up")