<s> """<STR_LIT>""" <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> from cStringIO import StringIO <EOL> import numpy as np <EOL> import h5py <EOL> import os <EOL> from cellprofiler . cpgridinfo import CPGridInfo <EOL> from . utilities . hdf5_dict import HDF5FileList , HDF5Dict <EOL> '''<STR_LIT>''' <EOL> DISPOSITION_CONTINUE = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> DISPOSITION_SKIP = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> DISPOSITION_PAUSE = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> DISPOSITION_CANCEL = "<STR_LIT>" <EOL> def is_workspace_file ( path ) : <EOL> '''<STR_LIT>''' <EOL> if not h5py . is_hdf5 ( path ) : <EOL> return False <EOL> h5file = h5py . File ( path , mode = "<STR_LIT:r>" ) <EOL> try : <EOL> if not HDF5FileList . has_file_list ( h5file ) : <EOL> return False <EOL> return HDF5Dict . has_hdf5_dict ( h5file ) <EOL> finally : <EOL> h5file . close ( ) <EOL> class Workspace ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> pipeline , <EOL> module , <EOL> image_set , <EOL> object_set , <EOL> measurements , <EOL> image_set_list , <EOL> frame = None , <EOL> create_new_window = False , <EOL> outlines = { } ) : <EOL> """<STR_LIT>""" <EOL> self . __pipeline = pipeline <EOL> self . __module = module <EOL> self . __image_set = image_set <EOL> self . __object_set = object_set <EOL> self . __measurements = measurements <EOL> self . __image_set_list = image_set_list <EOL> self . __frame = frame <EOL> self . __do_show = frame is not None <EOL> self . __outlines = outlines <EOL> self . __windows_used = [ ] <EOL> self . __create_new_window = create_new_window <EOL> self . __grid = { } <EOL> self . __disposition = DISPOSITION_CONTINUE <EOL> self . __disposition_listeners = [ ] <EOL> self . __in_background = False <EOL> self . __filename = None <EOL> self . __file_list = None <EOL> self . __loading = False <EOL> if measurements is not None : <EOL> self . set_file_list ( HDF5FileList ( measurements . hdf5_dict . hdf5_file ) ) <EOL> self . __notification_callbacks = [ ] <EOL> self . interaction_handler = None <EOL> self . post_run_display_handler = None <EOL> self . post_group_display_handler = None <EOL> self . cancel_handler = None <EOL> class DisplayData ( object ) : <EOL> pass <EOL> self . display_data = DisplayData ( ) <EOL> """<STR_LIT>""" <EOL> def refresh ( self ) : <EOL> """<STR_LIT>""" <EOL> for window in self . __windows_used : <EOL> window . figure . canvas . draw ( ) <EOL> def get_windows_used ( self ) : <EOL> return self . __windows_used <EOL> def get_pipeline ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __pipeline <EOL> pipeline = property ( get_pipeline ) <EOL> def get_image_set ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __image_set <EOL> def set_image_set_for_testing_only ( self , image_set_number ) : <EOL> self . __image_set = self . image_set_list . get_image_set ( image_set_number ) <EOL> image_set = property ( get_image_set ) <EOL> def get_image_set_list ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __image_set_list <EOL> image_set_list = property ( get_image_set_list ) <EOL> def get_object_set ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __object_set <EOL> object_set = property ( get_object_set ) <EOL> def get_objects ( self , objects_name ) : <EOL> """<STR_LIT>""" <EOL> return self . object_set . get_objects ( objects_name ) <EOL> def get_measurements ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __measurements <EOL> measurements = property ( get_measurements ) <EOL> def add_measurement ( self , object_name , feature_name , data ) : <EOL> """<STR_LIT>""" <EOL> self . measurements . add_measurement ( object_name , feature_name , data ) <EOL> def get_file_list ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . __file_list <EOL> def set_file_list ( self , file_list ) : <EOL> """<STR_LIT>""" <EOL> if self . __file_list is not None : <EOL> self . __file_list . remove_notification_callback ( self . __on_file_list_changed ) <EOL> self . __file_list = file_list <EOL> if file_list is not None : <EOL> self . __file_list . add_notification_callback ( self . __on_file_list_changed ) <EOL> file_list = property ( get_file_list ) <EOL> def get_grid ( self , grid_name ) : <EOL> '''<STR_LIT>''' <EOL> if not self . __grid . has_key ( grid_name ) : <EOL> raise ValueError ( "<STR_LIT>" % grid_name ) <EOL> return self . __grid [ grid_name ] <EOL> def set_grids ( self , last = None ) : <EOL> '''<STR_LIT>''' <EOL> if last is None : <EOL> last = { } <EOL> self . __grid = last <EOL> return self . __grid <EOL> def set_grid ( self , grid_name , grid_info ) : <EOL> '''<STR_LIT>''' <EOL> self . __grid [ grid_name ] = grid_info <EOL> def cache ( self ) : <EOL> self . measurements . cache ( ) <EOL> self . measurements . cache_object_set ( self . object_set ) <EOL> def get_frame ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __do_show : <EOL> return self . __frame <EOL> return None <EOL> frame = property ( get_frame ) <EOL> def show_frame ( self , do_show ) : <EOL> self . __do_show = do_show <EOL> def get_display ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __frame is not None <EOL> display = property ( get_display ) <EOL> def get_in_background ( self ) : <EOL> return self . __in_background <EOL> def set_in_background ( self , val ) : <EOL> self . __in_background = val <EOL> in_background = property ( get_in_background , set_in_background ) <EOL> def get_module_figure ( self , module , image_set_number , parent = None ) : <EOL> """<STR_LIT>""" <EOL> import cellprofiler . gui . cpfigure as cpf <EOL> import cellprofiler . measurements as cpmeas <EOL> assert not self . __in_background <EOL> window_name = cpf . window_name ( module ) <EOL> if self . measurements . has_feature ( cpmeas . EXPERIMENT , <EOL> cpmeas . M_GROUPING_TAGS ) : <EOL> group_number = self . measurements [ <EOL> cpmeas . IMAGE , cpmeas . GROUP_NUMBER , image_set_number ] <EOL> group_index = self . measurements [ <EOL> cpmeas . IMAGE , cpmeas . GROUP_INDEX , image_set_number ] <EOL> title = "<STR_LIT>" % ( <EOL> module . module_name , module . module_num , image_set_number , <EOL> group_number , group_index ) <EOL> else : <EOL> title = "<STR_LIT>" % ( module . module_name , <EOL> module . module_num , <EOL> image_set_number ) <EOL> if self . __create_new_window : <EOL> figure = cpf . CPFigureFrame ( parent or self . __frame , <EOL> name = window_name , <EOL> title = title ) <EOL> else : <EOL> figure = cpf . create_or_find ( parent or self . __frame , <EOL> name = window_name , <EOL> title = title ) <EOL> if not figure in self . __windows_used : <EOL> self . __windows_used . append ( figure ) <EOL> return figure <EOL> def create_or_find_figure ( self , title = None , subplots = None , window_name = None ) : <EOL> """<STR_LIT>""" <EOL> import cellprofiler . gui . cpfigure as cpf <EOL> assert not self . __in_background <EOL> if title is None : <EOL> title = self . __module . module_name <EOL> if window_name is None : <EOL> window_name = cpf . window_name ( self . __module ) <EOL> if self . __create_new_window : <EOL> figure = cpf . CPFigureFrame ( self , <EOL> title = title , <EOL> name = window_name , <EOL> subplots = subplots ) <EOL> else : <EOL> figure = cpf . create_or_find ( self . __frame , title = title , <EOL> name = window_name , <EOL> subplots = subplots ) <EOL> if not figure in self . __windows_used : <EOL> self . __windows_used . append ( figure ) <EOL> return figure <EOL> def get_outline_names ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __outlines . keys ( ) <EOL> def add_outline ( self , name , outline ) : <EOL> """<STR_LIT>""" <EOL> self . __outlines [ name ] = outline <EOL> def get_outline ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . __outlines [ name ] <EOL> def get_module ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __module <EOL> module = property ( get_module ) <EOL> def set_module ( self , module ) : <EOL> """<STR_LIT>""" <EOL> self . __module = module <EOL> def interaction_request ( self , module , * args , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> import cellprofiler . preferences as cpprefs <EOL> if "<STR_LIT>" in kwargs : <EOL> tmp = kwargs . copy ( ) <EOL> del tmp [ "<STR_LIT>" ] <EOL> headless_ok = kwargs [ "<STR_LIT>" ] <EOL> kwargs = tmp <EOL> else : <EOL> headless_ok = False <EOL> if self . interaction_handler is None : <EOL> if cpprefs . get_headless ( ) and not headless_ok : <EOL> raise self . NoInteractionException ( ) <EOL> else : <EOL> return module . handle_interaction ( * args , ** kwargs ) <EOL> else : <EOL> return self . interaction_handler ( module , * args , ** kwargs ) <EOL> def cancel_request ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . cancel_handler is None : <EOL> raise self . NoInteractionException ( ) <EOL> self . cancel_handler ( ) <EOL> def post_group_display ( self , module ) : <EOL> '''<STR_LIT>''' <EOL> if self . post_group_display_handler is not None : <EOL> self . post_group_display_handler ( <EOL> module , self . display_data , self . measurements . image_set_number ) <EOL> elif self . frame is not None : <EOL> figure = self . get_module_figure ( <EOL> module , <EOL> self . measurements . image_set_number , <EOL> self . frame ) <EOL> module . display_post_group ( self , figure ) <EOL> def post_run_display ( self , module ) : <EOL> '''<STR_LIT>''' <EOL> if self . post_run_display_handler is not None : <EOL> self . post_run_display_handler ( self , module ) <EOL> elif self . frame is not None : <EOL> figure = self . get_module_figure ( <EOL> module , <EOL> self . measurements . image_set_count + <NUM_LIT:1> , <EOL> self . frame ) <EOL> module . display_post_run ( self , figure ) <EOL> @ property <EOL> def is_last_image_set ( self ) : <EOL> return ( self . measurements . image_set_number == <EOL> self . image_set_list . count ( ) - <NUM_LIT:1> ) <EOL> def get_disposition ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . __disposition <EOL> def set_disposition ( self , disposition ) : <EOL> self . __disposition = disposition <EOL> event = DispositionChangedEvent ( disposition ) <EOL> for listener in self . __disposition_listeners : <EOL> listener ( event ) <EOL> disposition = property ( get_disposition , set_disposition ) <EOL> def add_disposition_listener ( self , listener ) : <EOL> self . __disposition_listeners . append ( listener ) <EOL> class NoInteractionException ( Exception ) : <EOL> pass <EOL> def load ( self , filename , load_pipeline ) : <EOL> '''<STR_LIT>''' <EOL> import shutil <EOL> from . pipeline import M_PIPELINE , M_DEFAULT_INPUT_FOLDER , M_DEFAULT_OUTPUT_FOLDER <EOL> import cellprofiler . measurements as cpmeas <EOL> from cellprofiler . preferences import set_default_image_directory , set_default_output_directory <EOL> image_set_and_measurements_are_same = False <EOL> if self . __measurements is not None : <EOL> image_set_and_measurements_are_same = ( <EOL> id ( self . __measurements ) == id ( self . __image_set ) ) <EOL> self . close ( ) <EOL> self . __loading = True <EOL> try : <EOL> fd , self . __filename = cpmeas . make_temporary_file ( ) <EOL> os . close ( fd ) <EOL> shutil . copyfile ( filename , self . __filename ) <EOL> self . __measurements = cpmeas . Measurements ( <EOL> filename = self . __filename , mode = "<STR_LIT>" ) <EOL> if self . __file_list is not None : <EOL> self . __file_list . remove_notification_callback ( <EOL> self . __on_file_list_changed ) <EOL> self . __file_list = HDF5FileList ( self . measurements . hdf5_dict . hdf5_file ) <EOL> self . __file_list . add_notification_callback ( self . __on_file_list_changed ) <EOL> if load_pipeline and self . __measurements . has_feature ( <EOL> cpmeas . EXPERIMENT , M_PIPELINE ) : <EOL> pipeline_txt = self . __measurements . get_experiment_measurement ( <EOL> M_PIPELINE ) . encode ( "<STR_LIT:utf-8>" ) <EOL> self . pipeline . load ( StringIO ( pipeline_txt ) ) <EOL> elif load_pipeline : <EOL> self . pipeline . clear ( ) <EOL> else : <EOL> fd = StringIO ( ) <EOL> self . pipeline . savetxt ( fd , save_image_plane_details = False ) <EOL> self . __measurements . add_experiment_measurement ( <EOL> M_PIPELINE , fd . getvalue ( ) ) <EOL> for feature , function in ( <EOL> ( M_DEFAULT_INPUT_FOLDER , set_default_image_directory ) , <EOL> ( M_DEFAULT_OUTPUT_FOLDER , set_default_output_directory ) ) : <EOL> if self . measurements . has_feature ( cpmeas . EXPERIMENT , feature ) : <EOL> path = self . measurements [ cpmeas . EXPERIMENT , feature ] <EOL> if os . path . isdir ( path ) : <EOL> function ( path ) <EOL> if image_set_and_measurements_are_same : <EOL> self . __image_set = self . __measurements <EOL> finally : <EOL> self . __loading = False <EOL> self . notify ( self . WorkspaceLoadedEvent ( self ) ) <EOL> def create ( self ) : <EOL> '''<STR_LIT>''' <EOL> from . measurements import Measurements , make_temporary_file <EOL> if isinstance ( self . measurements , Measurements ) : <EOL> self . close ( ) <EOL> fd , self . __filename = make_temporary_file ( ) <EOL> self . __measurements = Measurements ( <EOL> filename = self . __filename , mode = "<STR_LIT:w>" ) <EOL> os . close ( fd ) <EOL> if self . __file_list is not None : <EOL> self . __file_list . remove_notification_callback ( <EOL> self . __on_file_list_changed ) <EOL> self . __file_list = HDF5FileList ( self . measurements . hdf5_dict . hdf5_file ) <EOL> self . __file_list . add_notification_callback ( self . __on_file_list_changed ) <EOL> self . notify ( self . WorkspaceCreatedEvent ( self ) ) <EOL> def save ( self , path ) : <EOL> '''<STR_LIT>''' <EOL> self . save_default_folders_to_measurements ( ) <EOL> self . measurements . flush ( ) <EOL> hdf5src = self . measurements . hdf5_dict . hdf5_file <EOL> hdf5dest = h5py . File ( path , mode = "<STR_LIT:w>" ) <EOL> for key in hdf5src : <EOL> obj = hdf5src [ key ] <EOL> if isinstance ( obj , h5py . Dataset ) : <EOL> hdf5dest [ key ] = obj . value <EOL> else : <EOL> hdf5src . copy ( hdf5src [ key ] , hdf5dest , key ) <EOL> for key in hdf5src . attrs : <EOL> hdf5dest . attrs [ key ] = hdf5src . attrs [ key ] <EOL> hdf5dest . close ( ) <EOL> def close ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . measurements is not None and self . __filename is not None : <EOL> self . measurements . close ( ) <EOL> os . unlink ( self . __filename ) <EOL> def save_pipeline_to_measurements ( self ) : <EOL> from cellprofiler . pipeline import M_PIPELINE <EOL> fd = StringIO ( ) <EOL> self . pipeline . savetxt ( fd , save_image_plane_details = False ) <EOL> self . measurements . add_experiment_measurement ( M_PIPELINE , fd . getvalue ( ) ) <EOL> self . measurements . flush ( ) <EOL> def save_default_folders_to_measurements ( self ) : <EOL> from cellprofiler . pipeline import M_DEFAULT_INPUT_FOLDER <EOL> from cellprofiler . pipeline import M_DEFAULT_OUTPUT_FOLDER <EOL> from cellprofiler . preferences import get_default_image_directory <EOL> from cellprofiler . preferences import get_default_output_directory <EOL> self . measurements . add_experiment_measurement ( <EOL> M_DEFAULT_INPUT_FOLDER , get_default_image_directory ( ) ) <EOL> self . measurements . add_experiment_measurement ( <EOL> M_DEFAULT_OUTPUT_FOLDER , get_default_output_directory ( ) ) <EOL> def invalidate_image_set ( self ) : <EOL> if not self . __loading : <EOL> self . measurements . clear ( ) <EOL> self . save_pipeline_to_measurements ( ) <EOL> def refresh_image_set ( self , force = False ) : <EOL> '''<STR_LIT>''' <EOL> import cellprofiler . measurements as cpmeas <EOL> if len ( self . measurements . get_image_numbers ( ) ) == <NUM_LIT:0> or force : <EOL> self . measurements . clear ( ) <EOL> self . save_pipeline_to_measurements ( ) <EOL> modules = self . pipeline . modules ( ) <EOL> stop_module = None <EOL> if len ( modules ) > <NUM_LIT:1> and not modules [ - <NUM_LIT:1> ] . is_load_module ( ) : <EOL> for module , next_module in zip ( <EOL> self . pipeline . modules ( ) [ : - <NUM_LIT:1> ] , <EOL> self . pipeline . modules ( ) [ <NUM_LIT:1> : ] ) : <EOL> if module . is_load_module ( ) : <EOL> stop_module = next_module <EOL> no_image_set_list = self . image_set_list is None <EOL> if no_image_set_list : <EOL> from cellprofiler . cpimage import ImageSetList <EOL> self . __image_set_list = ImageSetList ( ) <EOL> try : <EOL> result = self . pipeline . prepare_run ( self , stop_module ) <EOL> return result <EOL> except : <EOL> logger . error ( "<STR_LIT>" , exc_info = <NUM_LIT:1> ) <EOL> return False <EOL> finally : <EOL> if no_image_set_list : <EOL> self . __image_set_list = None <EOL> return True <EOL> def add_notification_callback ( self , callback ) : <EOL> '''<STR_LIT>''' <EOL> self . __notification_callbacks . append ( callback ) <EOL> def remove_notification_callback ( self , callback ) : <EOL> self . __notification_callbacks . remove ( callback ) <EOL> def notify ( self , event ) : <EOL> for callback in self . __notification_callbacks : <EOL> try : <EOL> callback ( event ) <EOL> except : <EOL> logger . error ( "<STR_LIT>" , <EOL> exc_info = <NUM_LIT:1> ) <EOL> def __on_file_list_changed ( self ) : <EOL> self . notify ( self . WorkspaceFileListNotification ( self ) ) <EOL> class WorkspaceEvent ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , workspace ) : <EOL> self . workspace = workspace <EOL> class WorkspaceLoadedEvent ( WorkspaceEvent ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , workspace ) : <EOL> super ( self . __class__ , self ) . __init__ ( workspace ) <EOL> class WorkspaceCreatedEvent ( WorkspaceEvent ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , workspace ) : <EOL> super ( self . __class__ , self ) . __init__ ( workspace ) <EOL> class WorkspaceFileListNotification ( WorkspaceEvent ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , workspace ) : <EOL> super ( self . __class__ , self ) . __init__ ( workspace ) <EOL> class DispositionChangedEvent ( object ) : <EOL> def __init__ ( self , disposition ) : <EOL> self . disposition = disposition </s>
