"""Test the LoadImages module

CellProfiler is distributed under the GNU General Public License.
See the accompanying file LICENSE for details.

Copyright (c) 2003-2009 Massachusetts Institute of Technology
Copyright (c) 2009-2015 Broad Institute
All rights reserved.

Please see the AUTHORS file for credits.

Website: http://www.cellprofiler.org
"""
import base64
import gc
import glob
import hashlib
import numpy as np
import os
import re
import unittest
import tempfile
import time
import urllib
import sys
import zlib
from StringIO import StringIO
import traceback

import cellprofiler.pipeline as cpp
import cellprofiler.cpmodule as CPM
import cellprofiler.modules.loadimages as LI
import cellprofiler.modules.tests as T
import cellprofiler.cpimage as I
import cellprofiler.objects as cpo
import cellprofiler.measurements as measurements
import cellprofiler.pipeline as P
import cellprofiler.workspace as W
from cellprofiler.modules.tests import \
     example_images_directory, maybe_download_example_images, \
     maybe_download_sbs, maybe_download_test_image, maybe_download_fly,\
     example_images_url
from cellprofiler.modules.namesandtypes import M_IMAGE_SET
import cellprofiler.preferences as cpprefs
from bioformats.omexml import PT_UINT8
from bioformats.formatreader import clear_image_reader_cache
from bioformats.formatwriter import write_image

IMAGE_NAME = "image"
ALT_IMAGE_NAME = "altimage"
OBJECTS_NAME = "objects"
OUTLINES_NAME = "outlines"

class ConvtesterMixin:
    '''Mixin class that supplies a generic legacy conversion tester method
    
    '''
    def convtester(self, pipeline_text, directory, fn_filter = (lambda x: True)):
        '''Test whether a converted pipeline yields the same output
        
        pipeline_text - the pipeline as a text file
        
        directory - the default input directory
        
        fn_filter - a function that returns True if a file should be included
                    in the workspace file list.
        '''
        cpprefs.set_default_image_directory(directory)
        pipeline = cpp.Pipeline()
        pipeline.load(StringIO(pipeline_text))
        def callback(caller, event):
            self.assertFalse(isinstance(event, cpp.RunExceptionEvent))
        pipeline.add_listener(callback)
        m = [m for m in pipeline.modules() if isinstance(m, LI.LoadImages)][0]
        m1 = measurements.Measurements(mode="memory")
        w1 = W.Workspace(pipeline, m, m1, None, m1, None)
        pipeline.prepare_run(w1)
        
        m2 = measurements.Measurements(mode="memory")
        w2 = W.Workspace(pipeline, m, m2, None, m2, None)
        urls = [LI.pathname2url(os.path.join(directory, filename))
                 for filename in os.listdir(directory)
                 if fn_filter(filename)]
        w2.file_list.add_files_to_filelist(urls)
        pipeline.add_urls(urls, False)
        pipeline.convert_legacy_input_modules()
        pipeline.prepare_run(w2)
        
        ff1 = m1.get_feature_names(measurements.IMAGE)
        ffexpected = [f.replace("IMAGE_FOR_", "") for f in ff1 
                      if not f.startswith(measurements.C_METADATA)]
        ff2 = [x for x in m2.get_feature_names(measurements.IMAGE)
               if not any([x.startswith(y) for y in (
               measurements.C_FRAME, measurements.C_SERIES,
               measurements.C_OBJECTS_FRAME, 
               measurements.C_OBJECTS_SERIES,
               measurements.C_CHANNEL,
               measurements.C_OBJECTS_CHANNEL,
               measurements.C_METADATA,
               M_IMAGE_SET
               )])]
        self.assertItemsEqual(ffexpected, ff2)
        for feature in ff1:
            if feature.startswith(measurements.C_METADATA):
                self.assertTrue(m2.has_feature(measurements.IMAGE, feature))
        ff1a = filter((lambda x:not x.startswith(measurements.C_METADATA)),
                      ff1)
        self.assertEqual(m1.image_set_count, m2.image_set_count)
        image_numbers = m1.get_image_numbers()
        #
        # Order images by URL
        #
        m_url1 = sorted(ff1, key=lambda f: f.replace("IMAGE_FOR_", ""))
        m_url2 = sorted(ff2)
        order1, order2 = [np.lexsort(
            [mm.get_measurement(measurements.IMAGE, f, image_numbers)
             for f in m_url]) for mm, m_url in ((m1, m_url1), (m2, m_url2))]
        image_numbers1 = image_numbers[order1]
        image_numbers2 = image_numbers[order2]
        for f1, f2 in zip(ff1a, ff2):
            if f1 in (measurements.GROUP_INDEX, measurements.GROUP_NUMBER,
                      measurements.IMAGE_NUMBER):
                continue
            v1 = m1.get_measurement(measurements.IMAGE, f1,
                                    image_set_number = image_numbers1)
            v2 = m2.get_measurement(measurements.IMAGE, f2,
                                    image_set_number = image_numbers2)
            if (f1.startswith(measurements.C_PATH_NAME) or
                f1.startswith(measurements.C_OBJECTS_PATH_NAME)):
                for p1, p2 in zip(v1, v2):
                    self.assertEqual(os.path.normcase(p1),
                                     os.path.normcase(p2))
            elif (f1.startswith(measurements.C_URL) or
                  f1.startswit
h(measurements.C_OBJECTS_URL)):
                for p1, p2 in zip(v1, v2):
                    self.assertEqual(
                        os.path.normcase(LI.url2pathname(p1.encode("utf-8"))),
                        os.path.normcase(LI.url2pathname(p2.encode("utf-8"))))
            else:        
                np.testing.assert_array_equal(v1, v2)

class testLoadImages(unittest.TestCase, ConvtesterMixin):
    @classmethod
    def setUpClass(cls):
        may
be_download_sbs()
        
    def setUp(self):
        self.directory = None
        
    def tearDown(self):
        clear_image_reader_cache()
        if self.directory is not None:
            try:
                for path in (os.path.sep.join((self.directory, "*","*")),
                             os.path.sep.join((self.directory, "*"))):
                    files = glob.glob(path)
                    for filename in files:
                        if os.path.isfile(filename):
                            os.remove(filename)
                        else:
                            os.rmdir(filename)
                os.rmdir(self.directory)
            except:
                sys.stderr.write("Failed during file delete / teardown\n")
                traceback.print_exc()
        
    def error_callback(self, calller, event):
        if isinstance(event, P.RunExceptionEvent):
            self.fail(event.error.message)

    def test_00_00init(self):
        x=LI.LoadImages()
    
    def test_00_01version(self):
        self.assertEqual(LI.LoadImages().variable_revision_number, 11,
                         "LoadImages' version number has changed")
    
    def test_01_01load_image_text_match(self):
        l=LI.LoadImages()
        l.match_method.value = LI.MS_EXACT_MATCH
        l.location.dir_choice = LI.ABSOLUTE_FOLDER_NAME
        l.location.custom_path =\
            os.path.join(T.example_images_directory(),"ExampleSBSImages")
        l.images[0].common_text.value = "1-01-A-01.tif"
        l.images[0].channels[0].image_name.value = "my_image"
        l.module_num = 1
        image_set_list = I.ImageSetList()
        pipeline = P.Pipeline()
        pipeline.add_listener(self.error_callback)
        pipeline.add_module(l)
        m = measurements.Measurements()
        workspace = W.Workspace(pipeline, l, None, None, m, image_set_list)
        l.prepare_run(workspace)
        image_numbers = m.get_image_numbers()
        self.assertEqual(len(image_numbers), 1, 
                         "Expected one image set in the list")
        l.prepare_group(workspace, (), [1])
        image_set = image_set_list.get_image_set(0)
        l.run(W.Workspace(pipeline, l, image_set, cpo.ObjectSet(),
                          m, image_set_list))
        self.assertEqual(len(image_set.get_names()),1)
        self.assertEqual(image_set.get_names()[0],"my_image")
        self.assertTrue(image_set.get_image("my_ima