import os
from pickle import UnpicklingError
import logging
import pandas as pd
from skimage.exposure import equalize_adapthist, rescale_intensity
from traits.api import Instance, List, Any, Tuple, Dict
from pyfibre.core.base_multi_image_analyser import (
BaseMultiImageAnalyser)
from pyfibre.model.objects.segments import (
FibreSegment, CellSegment)
from pyfibre.model.objects.fibre_network import FibreNetwork
from pyfibre.io.object_io import (
save_fibre_networks, load_fibre_networks,
save_fibre_segments, load_fibre_segments,
save_cell_segments, load_cell_segments)
from pyfibre.io.network_io import save_network, load_network
from pyfibre.io.database_io import save_database, load_database
from pyfibre.model.tools.network_extraction import (
build_network, fibre_network_assignment
)
from pyfibre.model.tools.preprocessing import nl_means
from pyfibre.utilities import flatten_list, log_time
from .metric_analysers import SHGMetricAnalyser
from .shg_image import SHGImage
logger = logging.getLogger(__name__)
[docs]class SHGAnalyser(BaseMultiImageAnalyser):
#: Reference to multi image under analysis
multi_image = Instance(SHGImage)
database_names = ['global', 'fibre', 'network', 'cell']
#: Parameters used for FIRE algorithm
fire_parameters = Dict()
#: Parameters used for segmentation
segment_parameters = Dict()
#: Reference to networkx Graph generated by analysis
_network = Any
#: Reference to FibreNetwork objects generated by analysis
_fibre_networks = List(FibreNetwork)
#: Reference to FibreSegment objects generated by analysis
_fibre_segments = List(FibreSegment)
#: Reference to CellSegment objects generated by analysis
_cell_segments = List(CellSegment)
#: Reference to metric DataFrames generated by analysis
_databases = Tuple()
def _fire_parameters_default(self):
return {
'nuc_thresh': 2,
'nuc_radius': 11,
'lmp_thresh': 0.15,
'angle_thresh': 70,
'r_thresh': 7
}
def _segment_parameters_default(self):
return {
'min_fibre_size': 100,
'min_fibre_frac': 0.1,
'min_cell_size': 200,
'min_cell_frac': 0.01,
}
@property
def data_path(self):
"""Path for analysis data files"""
return os.path.join(self.analysis_path, 'data')
@property
def fig_path(self):
"""Path for figures generated by analysis"""
return os.path.join(self.analysis_path, 'fig')
@property
def _data_file(self):
return os.path.join(self.data_path, self.multi_image.name)
@property
def _fig_file(self):
return os.path.join(self.fig_path, self.multi_image.name)
def _figures_kwargs(self):
"""Generates keyword arguments for multi image create_figures
method"""
kwargs = {}
fibres = [
fibre_network.fibres for fibre_network in self._fibre_networks]
kwargs['network_graphs'] = [
fibre_network.graph for fibre_network in self._fibre_networks]
kwargs['fibre_graphs'] = [
fibre.graph for fibre in flatten_list(fibres)]
kwargs['fibre_regions'] = [
fibre_segment.region for fibre_segment in self._fibre_segments]
return kwargs
def _clear_attr(self):
"""Clears all private temporary attributes"""
self._network = None
self._fibre_networks = []
self._fibre_segments = []
self._cell_segments = []
self._databases = (None, None, None, None)
def _save_networks(self):
"""Save networkx Graphs representing fibre networks"""
save_network(self._network, self._data_file, "network")
save_fibre_networks(self._fibre_networks, self._data_file)
def _load_networks(self):
"""Load networkx Graphs representing fibre network"""
self._network = load_network(self._data_file, "network")
self._fibre_networks = load_fibre_networks(self._data_file)
def _save_segments(self):
"""Save FibreSegment and CellSegment instances
created during the analysis"""
save_fibre_segments(
self._fibre_segments, self._data_file,
shape=self.multi_image.shape)
save_cell_segments(
self._cell_segments, self._data_file,
shape=self.multi_image.shape)
def _load_segments(self):
"""Load FibreSegment and CellSegment instances
created during the analysis"""
self._fibre_segments = load_fibre_segments(
self._data_file, intensity_image=self.multi_image.shg_image)
self._cell_segments = load_cell_segments(
self._data_file, intensity_image=self.multi_image.shg_image)
def _load_databases(self):
"""Shadows public API to assign pandas DataFrames to private
attribute"""
self._databases = self.load_databases()
def _save_databases(self):
"""Shadows public API to save private pandas DataFrames attribute"""
self.save_databases(self._databases)
[docs] def save_databases(self, databases):
"""Save pandas DataFrame instances created during the analysis"""
for index, name in enumerate(self.database_names):
save_database(databases[index], self._data_file, f'{name}_metric')
[docs] def load_databases(self):
"""Load pandas DataFrame instances created during the analysis"""
databases = [
load_database(self._data_file, f'{name}_metric')
for name in self.database_names
]
return tuple(databases)
[docs] def make_directories(self):
"""Creates additional directories for analysis"""
super(SHGAnalyser, self).make_directories()
if not os.path.exists(self.data_path):
os.mkdir(self.data_path)
if not os.path.exists(self.fig_path):
os.mkdir(self.fig_path)
[docs] def get_analysis_options(self, runner):
"""Get image-specific options for analysis"""
network = runner.ow_network
segment = network or runner.ow_segment
metric = segment or runner.ow_metric
try:
self._load_networks()
except Exception:
logger.info(
f"Cannot load networks in {self.multi_image.name}")
network = True
segment = True
metric = True
try:
self._load_segments()
except Exception:
logger.info(
f"Cannot load segments in {self.multi_image.name}")
segment = True
metric = True
try:
self._load_databases()
except (UnpicklingError, Exception):
logger.info(
f"Cannot load metrics for {self.multi_image.name}")
metric = True
logger.debug("Analysis options:\n "
f"Extract Network = {network}\n "
f"Segment Image = {segment}\n "
f"Generate Metrics = {metric}\n "
f"Save Figures = {runner.save_figures}")
return network, segment, metric
[docs] @log_time(message='NETWORK EXTRACTION')
def network_analysis(
self, sigma, alpha, scale, p_denoise):
"""Perform FIRE algorithm on image and save networkx
objects for further analysis
Parameters
----------
sigma: float
Gaussian standard deviation to filter distance image
alpha: float
Alpha metric to use in hysteresis threshold algorithm
scale: float
Scaling factor to apply to image before performing algorithm
p_denoise: tuple (float); shape=(2,)
Parameters for non-linear means denoise algorithm
(used to remove noise)
"""
logger.debug("Applying AHE to SHG image")
norm_stack = self.multi_image.preprocess_images()
image_equal = equalize_adapthist(
rescale_intensity(norm_stack[0]))
logger.debug(
"Performing NL Denoise using local windows {} {}".format(
*p_denoise)
)
image_nl = nl_means(image_equal, p_denoise=p_denoise)
# Call FIRE algorithm to extract full image network
logger.debug(
"Calling FIRE algorithm using "
f"image scale {scale} "
f"alpha {alpha}"
)
self._network = build_network(
image_nl,
scale=scale,
sigma=sigma,
alpha=alpha,
**self.fire_parameters)
self._fibre_networks = fibre_network_assignment(self._network)
[docs] @log_time(message='SEGMENTATION')
def segmentation_analysis(self, scale):
"""Segment image into regions
Parameters
----------
scale: float
segment_parameters: Dict
"""
logger.debug("Segmenting Fibre and Cell regions")
self._fibre_segments, self._cell_segments = (
self.multi_image.segmentation_algorithm(
self._fibre_networks,
scale=scale,
**self.segment_parameters
)
)
[docs] @log_time(message='SHG METRICS')
def create_metrics(self, sigma):
"""Perform metric analysis on segmented image
Parameters
----------
sigma: float
"""
global_dataframe = pd.Series(dtype=object)
global_dataframe['File'] = self.multi_image.name
logger.debug(" Performing SHG Image analysis")
metric_analyser = SHGMetricAnalyser(
filename=self.multi_image.name,
image=self.multi_image.shg_image,
sigma=sigma,
networks=self._fibre_networks,
segments=self._fibre_segments
)
(segment_merics,
network_metrics,
global_metrics) = metric_analyser.analyse()
global_dataframe = global_dataframe.append(
global_metrics, ignore_index=False)
logger.debug(" Fibre segment analysis complete")
self._databases = tuple(
[global_dataframe, segment_merics, network_metrics, None]
)
[docs] @log_time(message='ANALYSIS')
def image_analysis(self, runner):
"""
Analyse input image by calculating metrics and
segmenting via FIRE algorithm
Parameters
----------
runner: PyFibreRunner
Instructions for all image analysis algorithms
Returns
-------
databases: list of pd.DataFrame
Metrics returned by this analysis for a single image
"""
network, segment, metric = self.get_analysis_options(
runner
)
self._clear_attr()
self.make_directories()
# Load or create list of FibreNetwork instances
if network:
self.network_analysis(
sigma=runner.sigma,
alpha=runner.alpha,
scale=runner.scale,
p_denoise=runner.p_denoise)
self._save_networks()
else:
self._load_networks()
# Load or create lists of FibreSegments
if segment:
self.segmentation_analysis(
scale=runner.scale)
self._save_segments()
else:
self._load_segments()
if metric:
self.create_metrics(sigma=runner.sigma)
self._save_databases()
else:
self._load_databases()
# Create figures
if runner.save_figures:
self.create_figures()
return self._databases