def test_list(self):
     """TestRegistry: Test list all register IFs."""
     registry = Registry()
     impact_functions = registry.list()
     expected = [
         'Polygon flood on buildings',
         'Polygon flood on roads',
         'Polygon flood on people',
         'Raster flood on population',
         'Raster flood on buildings',
         'Raster flood on roads (QGIS)',
         'Raster flood on roads (GDAL)',
         'Tsunami evacuation',
         'Classified raster hazard on buildings',
         'Classified raster hazard on population',
         'Continuous raster hazard on population',
         'Classified polygon hazard on population',
         'Classified polygon hazard on buildings',
         'Earthquake on buildings',
         'Earthquake ITB fatality function',
         'Earthquake PAGER fatality function',
         'Point volcano on buildings',
         'Polygon volcano on buildings',
         'Point volcano on population',
         'Polygon volcano on population']
     self.assertTrue(len(impact_functions) == len(expected))
     self.assertItemsEqual(expected, impact_functions)

    def test_filter_by_keywords_dev(self):
        """TestRegistry: Test filtering IF using hazard n exposure keywords.

        Note (IS): I use this test to check the result of IF filtering only.
        """
        registry = Registry()

        # Using keywords string
        hazard_keywords = {
            'continuous_hazard_unit': 'metres',
            'hazard': 'flood',
            'hazard_category': 'single_event',
            'layer_mode': 'continuous',
            'layer_purpose': 'hazard',
            'title': 'Jakarta flood like 2007 with structural improvements'
        }
        exposure_keywords = {
            'exposure': 'population',
            'exposure_unit': 'count',
            'layer_geometry': 'raster',
            'layer_mode': 'continuous',
            'layer_purpose': 'exposure',
            'title': 'Population'
        }

        impact_functions = registry.filter_by_keyword_string(
            hazard_keywords, exposure_keywords)

        print len(impact_functions)

        for i in impact_functions:
            print i.__name__

 def test_get_impact_function_instance(self):
     """TestRegistry: Test we can get an impact function instance."""
     # Getting an IF instance using its class name
     registry = Registry()
     class_name = 'FloodPolygonBuildingFunction'
     impact_function = registry.get_instance(class_name)
     result = impact_function.__class__.__name__
     message = 'Expecting FloodPolygonBuildingFunction. Got %s ' \
               'instead.' % result
     self.assertEqual(class_name, result, message)

    def test_filter_by_exposure_metadata(self):
        """TestRegistry: Test filtering IF by exposure metadata."""
        # Full metadata
        exposure_metadata = {
            'layer_mode': layer_mode_classified,
            'layer_geometry': layer_geometry_point,
            'exposure': exposure_structure,
            'exposure_unit': []
            }
        registry = Registry()
        impact_functions = registry.filter_by_exposure(
            registry.impact_functions, exposure_metadata)
        expected = [
            FloodPolygonBuildingFunction,
            FloodRasterBuildingFunction,
            TsunamiRasterBuildingFunction,
            ClassifiedRasterHazardBuildingFunction,
            ClassifiedPolygonHazardBuildingFunction,
            EarthquakeBuildingFunction,
            VolcanoPointBuildingFunction,
            VolcanoPolygonBuildingFunction
        ]
        message = 'Expecting \n%s.\n\nGot \n%s \n instead' % (
            '\n'.join([x.__name__ for x in expected]),
            '\n'.join([x.__name__ for x in impact_functions]))
        self.assertItemsEqual(expected, impact_functions, message)

        # Full metadata
        exposure_metadata = {
            'layer_mode': layer_mode_classified,
            'layer_geometry': layer_geometry_polygon,
            'exposure': exposure_structure,
            # 'exposure_unit': []
        }
        registry = Registry()
        impact_functions = registry.filter_by_exposure(
            registry.impact_functions, exposure_metadata)
        expected = [
            FloodPolygonBuildingFunction,
            FloodRasterBuildingFunction,
            TsunamiRasterBuildingFunction,
            ClassifiedRasterHazardBuildingFunction,
            ClassifiedPolygonHazardBuildingFunction,
            EarthquakeBuildingFunction,
            VolcanoPointBuildingFunction,
            VolcanoPolygonBuildingFunction,
        ]
        message = 'Expecting \n%s.\n\nGot \n%s instead' % (
            '\n'.join([x.__name__ for x in expected]),
            '\n'.join([x.__name__ for x in impact_functions]))
        self.assertItemsEqual(expected, impact_functions, message)

    def test_get_impact_function_class(self):
        """TestRegistry: Test we can get an impact function class."""
        # Getting an IF class using its class name
        registry = Registry()
        expected = 'FloodPolygonBuildingFunction'
        impact_function = registry.get_class(expected)

        # Check that it should be a class, not an instance
        message = 'Expecting a class, not an object'
        self.assertTrue(inspect.isclass(impact_function), message)

        # Check the IF name the same
        result = impact_function.__name__
        message = 'Expecting %s. Got %s instead.' % (expected, result)
        self.assertEqual(expected, result, message)

    def test_get_impact_functions_by_metadata(self):
        """TestRegistry: Test getting the impact functions by its metadata."""
        # Test getting the impact functions by 'id'
        registry = Registry()
        impact_function_id = 'FloodPolygonBuildingFunction'
        result = registry.filter_by_metadata('id', impact_function_id)
        expected = [FloodPolygonBuildingFunction]
        message = 'Expecting %s. Got %s instead' % (expected, result)
        self.assertEqual(expected, result, message)

        # Test getting the impact functions by 'name'
        impact_function_name = 'Polygon flood on buildings'
        result = registry.filter_by_metadata('name', impact_function_name)
        expected = [FloodPolygonBuildingFunction]
        message = 'Expecting %s. Got %s instead.' % (expected, result)
        self.assertEqual(expected, result, message)

    def test_filter_by_metadata(self):
        """TestRegistry: Test filtering IF by hazard and exposure metadata."""
        hazard_metadata = {
            'subcategory': hazard_flood,
            'units': unit_wetdry,
            'layer_constraints': layer_vector_polygon
        }

        exposure_metadata = {
            'subcategory': exposure_structure,
            'units': unit_building_type_type,
            'layer_constraints': layer_vector_polygon
        }

        registry = Registry()
        impact_functions = registry.filter(hazard_metadata, exposure_metadata)
        expected = [FloodPolygonBuildingFunction]
        message = 'Expecting %s. Got %s instead' % (expected, impact_functions)
        self.assertEqual(expected, impact_functions, message)

    def test_filter_by_hazard_metadata(self):
        """TestRegistry: Test filtering IF by hazard metadata."""
        # Full metadata
        hazard_metadata = {
            'layer_mode': layer_mode_continuous,
            'layer_geometry': layer_geometry_raster,
            'hazard_category': hazard_category_single_event,
            'hazard': hazard_earthquake,
            'continuous_hazard_unit': unit_mmi,
        }
        registry = Registry()
        impact_functions = registry.filter_by_hazard(
            registry.impact_functions, hazard_metadata)
        expected = [
            ITBFatalityFunction,
            EarthquakeBuildingFunction,
            PAGFatalityFunction,
            ITBBayesianFatalityFunction
        ]

        message = 'Expecting \n%s.\n\nGot \n%s instead' % (
            '\n'.join([x.__name__ for x in expected]),
            '\n'.join([x.__name__ for x in impact_functions]))
        self.assertItemsEqual(expected, impact_functions, message)

        # Miss one metadata
        hazard_metadata = {
            'layer_mode': layer_mode_continuous,
            'layer_geometry': layer_geometry_raster,
            'hazard_category': hazard_category_single_event,
            'hazard': hazard_earthquake,
            # 'continuous_hazard_unit': unit_mmi,
            }
        registry = Registry()
        impact_functions = registry.filter_by_hazard(
            registry.impact_functions, hazard_metadata)
        expected = [
            # ITBFatalityFunction,
            # EarthquakeBuildingFunction,
            # PAGFatalityFunction
        ]
        message = 'Expecting %s. Got %s instead' % (expected, impact_functions)
        self.assertItemsEqual(expected, impact_functions, message)

    def test_filter_by_keywords(self):
        """TestRegistry: Test filtering IF using hazard n exposure keywords."""
        # Using keywords string
        hazard_keywords = {
            'subcategory': 'flood',
            'units': 'wetdry',
            'layer_type': 'vector',
            'data_type': 'polygon'
        }

        exposure_keywords = {
            'subcategory': 'structure',
            'units': 'building_type',
            'layer_type': 'vector',
            'data_type': 'polygon'
        }
        registry = Registry()
        impact_functions = registry.filter_by_keyword_string(
            hazard_keywords, exposure_keywords)
        message = 'Registry should returns matched impact functions. ' \
                  'Nothing returned instead. Please check registered IF.'
        self.assertTrue(len(impact_functions) > 0, message)

        for impact_function in impact_functions:
            result = impact_function.metadata().as_dict()[
                'categories']['hazard']['subcategories']
            result_list = [subcat.get('id') for subcat in result]
            expected = 'flood'
            message = 'Expecting flood hazard impact functions. Got %s ' \
                      'instead' % result_list[0]
            self.assertTrue(expected in result_list, message)

            result = impact_function.metadata().as_dict()[
                'categories']['exposure']['subcategories']
            result_list = [subcat.get('id') for subcat in result]
            expected = 'structure'
            message = 'Expecting structure exposure impact functions. ' \
                      'Got %s instead' % result_list[0]
            self.assertTrue(expected in result_list, message)

    def test_filter_by_metadata(self):
        """TestRegistry: Test filtering IF by hazard and exposure metadata."""
        hazard_metadata = {
            'layer_mode': layer_mode_continuous,
            'layer_geometry': layer_geometry_raster,
            'hazard_category': hazard_category_single_event,
            'hazard': hazard_earthquake,
            'continuous_hazard_unit': unit_mmi
        }

        exposure_metadata = {
            'layer_mode': layer_mode_classified,
            'layer_geometry': layer_geometry_point,
            'exposure': exposure_structure,
        }

        registry = Registry()
        impact_functions = registry.filter(hazard_metadata, exposure_metadata)
        expected = [EarthquakeBuildingFunction]
        message = 'Expecting \n%s.\n\nGot \n%s instead' % (
            '\n'.join([x.__name__ for x in expected]),
            '\n'.join([x.__name__ for x in impact_functions]))
        self.assertEqual(expected, impact_functions, message)

    def test_register_and_clear(self):
        """TestRegistry: Test register and clear impact function."""
        registry = Registry()
        registry.clear()
        message = 'Expecting registry should be cleared. %s impact ' \
                  'functions exists instead' % len(registry.impact_functions)
        self.assertEqual(0, len(registry.impact_functions), message)

        registry.register(FloodPolygonBuildingFunction)
        message = 'Expecting registry will contains 1 impact functions. %s ' \
                  'impact functions exists' % len(registry.impact_functions)
        self.assertEqual(1, len(registry.impact_functions), message)

        result = registry.get_instance('FloodPolygonBuildingFunction')\
            .metadata().as_dict()['id']
        expected = 'FloodPolygonBuildingFunction'
        message = 'Expected registered impact function ID should be %s. ' \
                  'Got %s instead' % (expected, result)
        self.assertEqual(expected, result, message)

    def __init__(self, parent=None, iface=None):
        """Constructor for Raster Reclassification to Vector Polygon.

        .. versionadded: 3.4

        :param parent: Optional widget to use as parent
        :type parent: QWidget

        :param iface: An instance of QGisInterface
        :type iface: QGisInterface
        """
        QDialog.__init__(self, parent)
        self.parent = parent
        self.setupUi(self)

        self.setWindowTitle(self.tr('Raster Reclassification'))

        self.iface = iface

        self.if_registry = Registry()
        self.keyword_io = KeywordIO()

        # populate raster input
        self.cbo_raster_input.clear()
        registry = QgsMapLayerRegistry.instance()
        # MapLayers returns a QMap<QString id, QgsMapLayer layer>
        layers = registry.mapLayers().values()
        for layer in layers:
            try:
                name = layer.name()
                source = layer.id()
                layer_purpose = self.keyword_io.read_keywords(
                    layer, 'layer_purpose')
                if (isinstance(layer, QgsRasterLayer) and
                            layer_purpose == 'hazard'):
                    add_ordered_combo_item(
                        self.cbo_raster_input, self.tr(name), source)

            except Exception as e:
                raise e

        # self.input_list_parameter = InputListParameter()
        # self.input_list_parameter.name = 'Thresholds'
        # self.input_list_parameter.description = (
        #     'List of thresholds of values used in reclassification.')
        # self.input_list_parameter.help_text = 'list of thresholds used'
        # self.input_list_parameter.maximum_item_count = 100
        # self.input_list_parameter.minimum_item_count = 1
        # self.input_list_parameter.element_type = float
        # self.input_list_parameter.value = [0.0, 1.0]
        # self.input_list_parameter.ordering = \
        #     InputListParameter.AscendingOrder
        # self.thresholds_widget = InputListParameterWidget(
        #     self.input_list_parameter)

        # self.threshold_editor.layout().addWidget(self.thresholds_widget)

        # Set up context help
        self.help_button = self.button_box.button(QtGui.QDialogButtonBox.Help)
        self.ok_button = self.button_box.button(QtGui.QDialogButtonBox.Ok)
        # self.cancel_button = self.button_box.button(
        #     QtGui.QDialogButtonBox.Cancel)
        self.button_box.accepted.connect(self.accept)
        self.button_box.rejected.connect(self.reject)
        # Allow toggling the help button
        self.help_button.setCheckable(True)
        self.help_button.toggled.connect(self.help_toggled)
        self.main_stacked_widget.setCurrentIndex(1)
        # adapt layer changed
        self.cbo_raster_input.currentIndexChanged.connect(self.raster_changed)
        self.raster_changed(self.cbo_raster_input.currentIndex())

class ImpactFunctionManager(object):
    """Class for managing metadata for all impact function.

    .. versionadded:: 2.1
    """

    def __init__(self):
        """Constructor."""
        # Singleton Registry to track all the registered Impact Functions
        self.registry = Registry()

    @property
    def impact_functions(self):
        """Return all registered impact functions."""
        return self.registry.impact_functions

    def get_instance(self, class_name):
        """Return an instance of an impact function given its class name.

        .. example::

            if_manager = ImpactFunctionManager()
            if_class_name = 'FloodBuildingImpactFunction'
            if =  if_manager.get_instance(if_class_name)

        :param class_name: The name of IF class.
        :type class_name: str

        :return: Impact function instance that matches the argument.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        return self.registry.get_instance(class_name)

    def get_class(self, class_name):
        """Return the class of an impact function given its class name.

        .. example::

            if_manager = ImpactFunctionManager()
            if_class_name = 'FloodBuildingImpactFunction'
            if_class =  if_manager.get_class(if_class_name)

        :param class_name: the name of IF class
        :type class_name: str

        :return: impact function class that matches the argument.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        return self.registry.get_class(class_name)

    def get(self, impact_function_id):
        """Return an instance of an impact function given its ID.

        This is a preferred way to get an instance of IF. IF should have a
        unique human readable ID in their metadata.

        .. example::

            if_manager = ImpactFunctionManager()
            if_id = 'FloodBuildingImpactFunction'
            if =  if_manager.get(if_id)

        :param impact_function_id: The ID of impact function in the metadata.
        :type impact_function_id: str

        :return An Impact function instance that has matched id.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        impact_functions = self.registry.filter_by_metadata(
            'id', impact_function_id)
        if len(impact_functions) == 0:
            raise Exception(
                'Impact function with ID: %s not found' % impact_function_id)
        elif len(impact_functions) > 1:
            raise Exception(
                'There are some Impact Functions that have the same ID: %s' %
                impact_function_id)
        return impact_functions[0].instance()

    def filter(self, hazard_metadata=None, exposure_metadata=None):
        """Get available impact functions from hazard and exposure metadata.

        Disabled impact function will not be loaded.

        .. example::

            if_manager = ImpactFunctionManager()
            hazard_metadata = {
                'subcategory': hazard_flood,
                'units': unit_wetdry,
                'layer_constraints': layer_vector_polygon
            }
            exposure_metadata = {
                'subcategory': exposure_structure,
                'units': unit_building_type_type,
                'layer_constraints': layer_vector_polygon
            }
            ifs =  if_manager.filter(hazard_metadata, exposure_metadata)

        :param hazard_metadata: The metadata of the hazard.
#.........这里部分代码省略.........

 def test_list(self):
     """TestRegistry: Test list all register IFs."""
     registry = Registry()
     impact_functions = registry.list()
     self.assertItemsEqual(EXPECTED_IF, impact_functions)

class ImpactFunctionManager(object):
    """Class for managing metadata for all impact function.

    .. versionadded:: 2.1
    """

    def __init__(self):
        """Constructor."""
        # Singleton Registry to track all the registered Impact Functions
        self.registry = Registry()

    @property
    def impact_functions(self):
        """Return all registered impact functions."""
        return self.registry.impact_functions

    def get_instance(self, class_name):
        """Return an instance of an impact function given its class name.

        .. example::

            if_manager = ImpactFunctionManager()
            if_class_name = 'FloodBuildingImpactFunction'
            if =  if_manager.get_instance(if_class_name)

        :param class_name: The name of IF class.
        :type class_name: str

        :return: Impact function instance that matches the argument.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        return self.registry.get_instance(class_name)

    def get_class(self, class_name):
        """Return the class of an impact function given its class name.

        .. example::

            if_manager = ImpactFunctionManager()
            if_class_name = 'FloodBuildingImpactFunction'
            if_class =  if_manager.get_class(if_class_name)

        :param class_name: the name of IF class
        :type class_name: str

        :return: impact function class that matches the argument.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        return self.registry.get_class(class_name)

    def get(self, impact_function_id):
        """Return an instance of an impact function given its ID.

        This is a preferred way to get an instance of IF. IF should have a
        unique human readable ID in their metadata.

        .. example::

            if_manager = ImpactFunctionManager()
            if_id = 'FloodBuildingImpactFunction'
            if =  if_manager.get(if_id)

        :param impact_function_id: The ID of impact function in the metadata.
        :type impact_function_id: str

        :return An Impact function instance that has matched id.
        :rtype: safe.impact_functions.base.ImpactFunction
        """
        impact_functions = self.registry.filter_by_metadata("id", impact_function_id)
        if len(impact_functions) == 0:
            raise Exception("Impact function with ID: %s not found" % impact_function_id)
        elif len(impact_functions) > 1:
            raise Exception("There are some Impact Functions that have the same ID: %s" % impact_function_id)
        return impact_functions[0].instance()

    def filter(self, hazard_metadata=None, exposure_metadata=None):
        """Get available impact functions from hazard and exposure metadata.

        Disabled impact function will not be loaded.

        .. example::

            if_manager = ImpactFunctionManager()
            hazard_metadata = {
                'subcategory': hazard_flood,
                'units': unit_wetdry,
                'layer_constraints': layer_vector_polygon
            }
            exposure_metadata = {
                'subcategory': exposure_structure,
                'units': unit_building_type_type,
                'layer_constraints': layer_vector_polygon
            }
            ifs =  if_manager.filter(hazard_metadata, exposure_metadata)

        :param hazard_metadata: The metadata of the hazard.
        :type hazard_metadata: dict

        :param exposure_metadata: The metadata of the exposure.
        :type exposure_metadata: dict
#.........这里部分代码省略.........

 def __init__(self):
     """Constructor."""
     # Singleton Registry to track all the registered Impact Functions
     self.registry = Registry()

class RasterReclassifyDialog(QDialog, FORM_CLASS):

    def __init__(self, parent=None, iface=None):
        """Constructor for Raster Reclassification to Vector Polygon.

        .. versionadded: 3.4

        :param parent: Optional widget to use as parent
        :type parent: QWidget

        :param iface: An instance of QGisInterface
        :type iface: QGisInterface
        """
        QDialog.__init__(self, parent)
        self.parent = parent
        self.setupUi(self)

        self.setWindowTitle(self.tr('Raster Reclassification'))

        self.iface = iface

        self.if_registry = Registry()
        self.keyword_io = KeywordIO()

        # populate raster input
        self.cbo_raster_input.clear()
        registry = QgsMapLayerRegistry.instance()
        # MapLayers returns a QMap<QString id, QgsMapLayer layer>
        layers = registry.mapLayers().values()
        for layer in layers:
            try:
                name = layer.name()
                source = layer.id()
                layer_purpose = self.keyword_io.read_keywords(
                    layer, 'layer_purpose')
                if (isinstance(layer, QgsRasterLayer) and
                            layer_purpose == 'hazard'):
                    add_ordered_combo_item(
                        self.cbo_raster_input, self.tr(name), source)

            except Exception as e:
                raise e

        # self.input_list_parameter = InputListParameter()
        # self.input_list_parameter.name = 'Thresholds'
        # self.input_list_parameter.description = (
        #     'List of thresholds of values used in reclassification.')
        # self.input_list_parameter.help_text = 'list of thresholds used'
        # self.input_list_parameter.maximum_item_count = 100
        # self.input_list_parameter.minimum_item_count = 1
        # self.input_list_parameter.element_type = float
        # self.input_list_parameter.value = [0.0, 1.0]
        # self.input_list_parameter.ordering = \
        #     InputListParameter.AscendingOrder
        # self.thresholds_widget = InputListParameterWidget(
        #     self.input_list_parameter)

        # self.threshold_editor.layout().addWidget(self.thresholds_widget)

        # Set up context help
        self.help_button = self.button_box.button(QtGui.QDialogButtonBox.Help)
        self.ok_button = self.button_box.button(QtGui.QDialogButtonBox.Ok)
        # self.cancel_button = self.button_box.button(
        #     QtGui.QDialogButtonBox.Cancel)
        self.button_box.accepted.connect(self.accept)
        self.button_box.rejected.connect(self.reject)
        # Allow toggling the help button
        self.help_button.setCheckable(True)
        self.help_button.toggled.connect(self.help_toggled)
        self.main_stacked_widget.setCurrentIndex(1)
        # adapt layer changed
        self.cbo_raster_input.currentIndexChanged.connect(self.raster_changed)
        self.raster_changed(self.cbo_raster_input.currentIndex())

    # noinspection PyPep8Naming
    @pyqtSlot(int)
    def raster_changed(self, index):
        """Executed when raster is changed

        :param index: index of the selected raster
        :return:
        """
        registry = QgsMapLayerRegistry.instance()
        layer_id = self.cbo_raster_input.itemData(
            index, QtCore.Qt.UserRole)
        layer = registry.mapLayer(layer_id)
        layer_purpose = self.keyword_io.read_keywords(layer, 'layer_purpose')
        if layer_purpose == 'hazard':
            impact_function = self.if_registry.filter_by_hazard(
                self.if_registry.impact_functions,
                self.keyword_io.read_keywords(layer)
            )
        elif layer_purpose == 'exposure':
            impact_function = self.if_registry.filter_by_exposure(
                self.if_registry.impact_functions,
                self.keyword_io.read_keywords(layer)
            )
        else:
            impact_function = []

#.........这里部分代码省略.........