text:
# Copyright 2020 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""
Quantization aware training

User can use quantization aware to train a model. MindSpore supports quantization aware training,
which models quantization errors in both the forward and backward passes using fake-quantization
operations. Note that the entire computation is carried out in floating point. At the end of quantization
aware training, MindSpore provides conversion functions to convert the trained model into lower precision.
"""

import re
import mindspore.context as context
import numpy as np
from ... import nn, ops
from ..._checkparam import Validator, Rel
from ...nn.layer import quant
from ...ops import functional as F
from ..common import QuantDtype
from .quantizer import Quantizer, OptimizeOption
from .quant_utils import compute_KL_threshold

__all__ = ["QuantizationAwareTraining", "create_quant_config"]

def create_quant_config(quant_observer=(nn.FakeQuantWithMinMaxObserver, nn.FakeQuantWithMinMaxObserver),
                        quant_delay=(0, 0),
                        quant_dtype=(QuantDtype.INT8, QuantDtype.INT8),
                        per_channel=(False, False),
                        symmetric=(False, False),
                        narrow_range=(False, False),
                        mode="DEFAULT"):
    r"""
    Config the observer type of weights and data flow with quant params.

    Args:
        quant_observer (Union[Observer, list, tuple]): The types of observer for quantization. The first element
            applies to weights and second applies to data flow. Currently, only
            :class:`FakeQuantWithMinMaxObserver` supported.
            Default: (nn.FakeQuantWithMinMaxObserver, nn.FakeQuantWithMinMaxObserver).
        quant_delay (Union[int, list, tuple]): Number of steps after which weights and activations are quantized
            during train and eval. The first element represents weights and second element represents data flow.
            Default: (0, 0).
        quant_dtype (Union[QuantDtype, list, tuple]): Datatype to use for quantize weights and activations. The first
            element represents weights and second element represents data flow.
            Default: (QuantDtype.INT8, QuantDtype.INT8).
        per_channel (Union[bool, list, tuple]):  Quantization granularity based on layer or on channel. If `True`
            then base on per channel otherwise base on per layer. The first element represents weights
            and second element represents data flow, and second element must be `False` now.
            Default: (False, False).
        symmetric (Union[bool, list, tuple]): Whether the quantization algorithm is symmetric or not. If `True` then
            base on symmetric otherwise base on asymmetric. The first element represents weights and second
            element represents data flow. Default: (False, False).
        narrow_range (Union[bool, list, tuple]): Whether the quantization algorithm uses narrow range or not.
            The first element represents weights and the second element represents data flow.
            Default: (False, False).
        mode (str): Optional quantization mode, currently only `DEFAULT`(QAT) and `LEARNED_SCALE` are supported.
            Default: ("DEFAULT").

    Returns:
        QuantConfig, contains the observer type of weight and activation.

    Raises:
        ValueError: If the second element of `per_channel` is not `False`.
    """
    if per_channel[-1]:
        raise ValueError("Arg 'per_channel' second element must be 'False'.")
    weight_observer = quant_observer[0].partial_init(quant_delay=quant_delay[0], quant_dtype=quant_dtype[0],
                                                     per_channel=per_channel[0], symmetric=symmetric[0],
                                                     narrow_range=narrow_range[0], mode=mode)
    act_observer = quant_observer[-1].partial_init(quant_delay=quant_delay[-1], quant_dtype=quant_dtype[-1],
                                                   per_channel=per_channel[-1], symmetric=symmetric[-1],
                                                   narrow_range=narrow_range[-1], mode=mode)
    return quant.QuantConfig(weight=weight_observer, activation=act_observer)

class _AddFakeQuantInput(nn.Cell):
    """
    Add FakeQuant OP at input of the network. Only support one input case.
    """

    def __init__(self, network, quant_delay=0):
        super(_AddFakeQuantInput, self).__init__(auto_prefix=False)
        self.fake_quant_input = quant.FakeQuantWithMinMaxObserver(min_init=-6, max_init=6,
                                                                  quant_delay=quant_delay, ema=True)
        self.fake_quant_input.update_parameters_name('fake_quant_input.')
        self.network = network

    def construct(self, data):
        data = self.fake_quant_input(data)
        output = self.network(data)
        return output

class _AddFakeQuantAfterSubCell(nn.Cell):
    """
    Add FakeQuant OP after of the sub Cell.
    """

    def __init__(self, subcell, **kwargs):
        super(_AddFakeQuantAfterSubCell, self).__init__(auto_prefix=False)
        self.subcell = subcell
        self.mode = "DEFAULT"
        self.max_init = 6
        self.min_init = -6

        if OptimizeOption.LEARNED_SCALE in kwargs["optimize_option"]:
            self.mode = "LEARNED_SCALE"
            self.max_init = 16
            self.min_init = -16

        self.fake_quant_act = quant.FakeQuantWithMinMaxObserver(min_init=self.min_init,
                                                                max_init=self.max_init,
                                                                ema=True,
                                                                quant_dtype=kwargs["quant_dtype"],
                                                                quant_delay=kwargs["quant_delay"],
                                                                per_channel=kwargs["per_channel"],
                                                                symmetric=kwargs["symmetric"],
                                                                narrow_range=kwargs["narrow_range"],
                                                                mode=self.mode)

    def construct(self, *data):
        output = self.subcell(*data)
        output = self.fake_quant_act(output)
        return output

class QuantizationAwareTraining(Quantizer):
    r"""
    Quantizer for quantization aware training.

    Args:
        bn_fold (bool): Whether to use bn fold ops for simulation inference operation. Default: True.
        freeze_bn (int): Number of steps after which BatchNorm OP parameters fixed to global mean and variance.
            Default: 1e7.
        quant_delay (Union[int, list, tuple]): Number of steps after which weights and activations are quantized
            during train and eval. The first element represents weights and second element represents data flow.
            Default: (0, 0).
        quant_dtype (Union[QuantDtype, list, tuple]): Datatype to use for quantize weights and activations. The first
            element represents weights and second element represents data flow. It is necessary to consider the
            precision support of hardware devices in the practical quantization infer scenario.
            Default: (QuantDtype.INT8, QuantDtype.INT8).
        per_channel (Union[bool, list, tuple]):  Quantization granularity based on layer or on channel. If `True`
            then base on per channel otherwise base on per layer. The first element represents weights and second
            element represents data flow, and second element must be `False` now. Default: (False, False).
        symmetric (Union[bool, list, tuple]): Whether the quantization algorithm is symmetric or not. If `True` then
            base on symmetric otherwise base on asymmetric. The first element represents weights and second
            element represents data flow. Default: (False, False).
        narrow_range (Union[bool, list, tuple]): Whether the quantization algorithm uses narrow range or not.
            The first element represents weights and the second element represents data flow.
            Default: (False, False).
        optimize_option (Union[OptimizeOption, list, tuple]): Specifies the quant algorithm and options, currently
            only support `QAT` and `LEARNED_SCALE` (Note that, if both `QAT` and `LEARNED_SCALE` are configured,
            `LEARNED_SCALE` has a higher priority. `LEARNED_SCALE` currently only work under some constraints, which
            includes: freeze_bn=0, quant_delay=0, symmetric=Ture, narrow_range=True, More specifically, for operators
            such as ReLu and ReLu6, which only have positive values, we add a negative truncation to optimize this
            scenario, and narrow_range will automatically match to False). Default: OptimizeOption.QAT.
        one_conv_fold (bool): Whether to use one conv bn fold ops for simulation inference operation. Default: True.

    Raises:
        TypeError: If the element of `quant_delay` or `freeze_bn` is not int.
        TypeError: If `bn_fold`, `one_conv_fold` or the element of `per_channel`, `symmetric`, `narrow_range`
            is not bool.
        TypeError: If the element of `quant_dtype` is not `QuantDtype`.
        ValueError: If the length of `quant_delay`, `quant_dtype`, `per_channel`, `symmetric` or `narrow_range` is
            not less than 2.
        ValueError: If the `optimize_option` is `LEARNED_SCALE` and `freeze_bn` is not equal to 0.
        ValueError: If the `optimize_option` is `LEARNED_SCALE` and `symmetric` is not (True, True).
        ValueError: If the `optimize_option` is `LEARNED_SCALE` and `narrow_range` is not (True, True).
        ValueError: If the `optimize_option` is `LEARNED_SCALE` and `quant_delay` is not (0, 0).

    Examples:
        >>> from mindspore.compression.quant import QuantizationAwareTraining
        >>> class LeNet5(nn.Cell):
        ...     def __init__(self, num_class=10, channel=1):
        ...         super(LeNet5, self).__init__()
        ...         self.type = "fusion"
        ...         self.num_class = num_class
        ...
        ...         # change `nn.Conv2d` to `nn.Conv2dBnAct`
        ...         self.conv1 = nn.Conv2dBnAct(channel, 6, 5, pad_mode='valid', activation='relu')
        ...         self.conv2 = nn.Conv2dBnAct(6, 16, 5, pad_mode='valid', activation='relu')
        ...         # change `nn.Dense` to `nn.DenseBnAct`
        ...         self.fc1 = nn.DenseBnAct(16 * 5 * 5, 120, activation='relu')
        ...         self.fc2 = nn.DenseBnAct(120, 84, activation='relu')
        ...         self.fc3 = nn.DenseBnAct(84, self.num_class)
        ...
        ...         self.max_pool2d = nn.MaxPool2d(kernel_size=2, stride=2)
        ...         self.flatten = nn.Flatten()
        ...
        ...     def construct(self, x):
        ...         x = self.conv1(x)
        ...         x = self.max_pool2d(x)
        ...         x = self.conv2(x)
        ...         x = self.max_pool2d(x)
        ...         x = self.flatten(x)
        ...         x = self.fc1(x)
        ...         x = self.fc2(x)
        ...         x = self.fc3(x)
        ...         return x
        ...
        >>> net = LeNet5()
        >>> quantizer = QuantizationAwareTraining(bn_fold=False, per_channel=[True, False], symmetric=[True, False])
        >>> net_qat = quantizer.quantize(net)
    """
    __quant_op_name__ = ["Add", "Sub", "Mul", "RealDiv", "ReduceMean"]

    def __init__(self,
                 bn_fold=True,
                 freeze_bn=10000000,
                 quant_delay=(0, 0),
                 quant_dtype=(QuantDtype.INT8, QuantDtype.INT8),
                 per_channel=(False, False),
                 symmetric=(False, False),
                 narrow_range=(False, False),
                 optimize_option=OptimizeOption.QAT,
                 one_conv_fold=True):
        """Init for QuantizationAwareTraining quantizer"""
        super(QuantizationAwareTraining, self).__init__(optimize_option=optimize_option)

        def convert2list(name, value):
            if not isinstance(value, list) and not isinstance(value, tuple):
                value = [value]
            elif len(value) > 2:
                raise ValueError("input `{}` len should less then 2".format(name))
            return value

        quant_delay = convert2list("quant delay", quant_delay)
        quant_dtype = convert2list("quant dtype", quant_dtype)
        per_channel = convert2list("per channel", per_channel)
        symmetric = convert2list("symmetric", symmetric)
        narrow_range = convert2list("narrow range", narrow_range)

        self.weight_qdelay = Validator.check_non_negative_int(quant_delay[0], "quant delay")
        self.act_qdelay = Validator.check_int(quant_delay[-1], 0, Rel.GE, "quant delay")
        self.bn_fold = Validator.check_bool(bn_fold, "bn fold")
        self.freeze_bn = Validator.check_non_negative_int(freeze_bn, "freeze bn")
        self.weight_dtype = Validator.check_isinstance("weights dtype", quant_dtype[0], QuantDtype)
        self.act_dtype = Validator.check_isinstance("activations dtype", quant_dtype[-1], QuantDtype)
        self.weight_channel = Validator.check_bool(per_channel[0], "per channel")
        self.act_channel = Validator.check_bool(per_channel[-1], "per channel")
        self.weight_symmetric = Validator.check_bool(symmetric[0], "symmetric")
        self.act_symmetric = Validator.check_bool(symmetric[-1], "symmetric")
        self.weight_range = Validator.check_bool(narrow_range[0], "narrow range")
        self.act_range = Validator.check_bool(narrow_range[-1], "narrow range")
        self.one_conv_fold = Validator.check_bool(one_conv_fold, "one conv fold")
        self._convert_method_map = {nn.Conv2dBnAct: self._convert_conv,
                                    nn.DenseBnAct: self._convert_dense}
        self.mode = "DEFAULT"
        if OptimizeOption.LEARNED_SCALE in self.optimize_option:
            self.mode = "LEARNED_SCALE"
            if not self.weight_symmetric or not self.act_symmetric:
                raise ValueError("OptimizeOption.LEARNED_SCALE currently only support "
                                 "symmetric=(True, True) for quant")
            if not self.weight_range or not self.act_range:
                raise ValueError("OptimizeOption.LEARNED_SCALE currently only support narrow_range=(True, True) "
                                 "for quant")
            if self.freeze_bn != 0:
                raise ValueError("OptimizeOption.LEARNED_SCALE currently only support freeze_bn equal to 0, "
                                 "but get freeze_bn={}".format(self.freeze_bn))
            if self.weight_qdelay != 0 or self.act_qdelay != 0:
                raise ValueError("OptimizeOption.LEARNED_SCALE currently only support quant_delay=(0, 0)")
        self.quant_config = create_quant_config(quant_delay=quant_delay,
                                                quant_dtype=quant_dtype,
                                                per_channel=per_channel,
                                                symmetric=symmetric,
                                                narrow_range=narrow_range,
                                                mode=self.mode)
        self.eps = 1e-5

    def _convert_op_name(self, name):
        pattern = re.compile(r'([A-Z]{1})')
        name_new = re.sub(pattern, r'_\1', name).lower()
        if name_new[0] == '_':
            name_new = name_new[1:]
        return name_new

    def quantize(self, network):
        """
        Quant API to convert input network to a quantization aware training network.

        Note:
            Please refer to the Examples of class: `mindspore.compression.quant.QuantizationAwareTraining`.

        Args:
            network (Cell): network to be quantized.

        Returns:
            Cell, a quantization aware training network.

        Raises:
            KeyError: If the `device_target` set in context is not in `support_device`.
        """
        support_device = ["Ascend", "GPU"]
        if context.get_context('device_target') not in support_device:
            raise KeyError("Unsupported {} device target.".format(context.get_context('device_target')))

        if OptimizeOption.QAT in self.optimize_option or OptimizeOption.LEARNED_SCALE in self.optimize_option:
            network.update_cell_prefix()
            network = self._convert_subcells2quant(network)
            network.update_cell_type("quant")
        return network

    def _convert_subcells2quant(self, network):
        """
        convert sub cell like `Conv2dBnAct` and `DenseBnAct` to quant cell
        """
        cells = network.name_cells()
        change = False
        for name in cells:
            subcell = cells[name]
            if subcell == network:
                continue
            elif isinstance(subcell, (nn.Conv2dBnAct, nn.DenseBnAct)):
                prefix = subcell.param_prefix
                new_subcell = self._convert_method_map[type(subcell)](subcell)
                new_subcell.update_parameters_name(prefix + '.')
                network.insert_child_to_cell(name, new_subcell)
                change = True
            else:
                self._convert_subcells2quant(subcell)
        if isinstance(network, nn.SequentialCell) and change:
            network.cell_list = list(network.cells())

        # add FakeQuant OP after OP in white list, but not including those wrapped in the below quantization cell.
        if isinstance(network, (nn.FakeQuantWithMinMaxObserver,
                                nn.Conv2dBnFoldQuantOneConv,
                                nn.Conv2dBnFoldQuant,
                                nn.Conv2dBnWithoutFoldQuant,
                                nn.Conv2dQuant,
                                nn.DenseQuant,
                                nn.ActQuant,
                                nn.TensorAddQuant,
                                nn.MulQuant)):
            return network

        add_list = []
        for name in network.__dict__:
            if name[0] == '_':
                continue
            attr = network.__dict__[name]
            if isinstance(attr, ops.Primitive) and attr.name in self.__quant_op_name__:
                add_list.append((name, attr))
        for name, prim_op in add_list:
            prefix = name
            add_quant = _AddFakeQuantAfterSubCell(prim_op,
                                                  quant_dtype=self.act_dtype,
                                                  quant_delay=self.act_qdelay,
                                                  per_channel=self.act_channel,
                                                  symmetric=self.act_symmetric,
                                                  narrow_range=self.act_range,
                                                  optimize_option=self.optimize_option)
            if network.param_prefix:
                prefix = '.'.join([network.param_prefix, prefix])
            add_quant.update_parameters_name(prefix + '.')
            del network.__dict__[name]
            network.insert_child_to_cell(name, add_quant)
        return network

    def _convert_conv(self, subcell):
        """
        convert Conv2d cell to quant cell
        """
        min_init = -6
        max_init = 6
        if OptimizeOption.LEARNED_SCALE in self.optimize_option:
            subcell_weight_para = subcell.conv.weight.data.asnumpy()
            if subcell.has_bn:
                scale_factor = (subcell.batchnorm.gamma.data.asnumpy() /
                                np.sqrt(subcell.batchnorm.moving_variance.data.asnumpy() + self.eps))
                subcell_weight_para = subcell_weight_para * scale_factor.reshape(-1, 1, 1, 1)
            min_init, max_init = self._KL_init(subcell_weight_para, self.weight_dtype)
        self.quant_config = self.quant_config._replace(
            weight=self.quant_config.weight.partial_init(min_init=min_init, max_init=max_init))

        conv_inner = subcell.conv
        if subcell.has_bn:
            bn_inner = subcell.batchnorm
            if self.bn_fold:
                if self.one_conv_fold:
                    conv_inner = quant.Conv2dBnFoldQuantOneConv(conv_inner.in_channels,
                                                                conv_inner.out_channels,
                                                                kernel_size=conv_inner.kernel_size,
                                                                stride=conv_inner.stride,
                                                                pad_mode=conv_inner.pad_mode,
                                                                padding=conv_inner.padding,
                                                                dilation=conv_inner.dilation,
                                                                group=conv_inner.group,
                                                                eps=bn_inner.eps,
                                                                momentum=1 - bn_inner.momentum,
                                                                has_bias=conv_inner.has_bias,
                                                                bias_init=conv_inner.bias_init,
                                                                quant_config=self.quant_config,
                                                                quant_dtype=self.weight_dtype,
                                                                fake=True)
                else:
                    conv_inner = quant.Conv2dBnFoldQuant(conv_inner.in_channels,
                                                         conv_inner.out_channels,
                                                         kernel_size=conv_inner.kernel_size,
                                                         stride=conv_inner.stride,
                                                         pad_mode=conv_inner.pad_mode,
                                                         padding=conv_inner.padding,
                                                         dilation=conv_inner.dilation,
                                                         group=conv_inner.group,
                                                         eps=bn_inner.eps,
                                                         momentum=1 - bn_inner.momentum,
                                                         has_bias=conv_inner.has_bias,
                                                         bias_init=conv_inner.bias_init,
                                                         freeze_bn=self.freeze_bn,
                                                         quant_config=self.quant_config,
                                                         quant_dtype=self.weight_dtype,
                                                         fake=True)
                # change original network Batch Normalization OP parameters to quant network
                conv_inner.gamma = subcell.batchnorm.gamma
                conv_inner.beta = subcell.batchnorm.beta
                conv_inner.moving_mean = subcell.batchnorm.moving_mean
                conv_inner.moving_variance = subcell.batchnorm.moving_variance
            else:
                conv_inner = quant.Conv2dBnWithoutFoldQuant(conv_inner.in_channels,
                                                            conv_inner.out_channels,
                                                            kernel_size=conv_inner.kernel_size,
                                                            stride=conv_inner.stride,
                                                            pad_mode=conv_inner.pad_mode,
                                                            padding=conv_inner.padding,
                                                            dilation=conv_inner.dilation,
                                                            group=conv_inner.group,
                                                            eps=bn_inner.eps,
                                                            momentum=1 - bn_inner.momentum,
                                                            has_bias=conv_inner.has_bias,
                                                            bias_init=conv_inner.bias_init,
                                                            quant_config=self.quant_config,
                                                            quant_dtype=self.weight_dtype)
                # change original network Batch Normalization OP parameters to quant network
                conv_inner.batchnorm.gamma = subcell.batchnorm.gamma
                conv_inner.batchnorm.beta = subcell.batchnorm.beta
                conv_inner.batchnorm.moving_mean = subcell.batchnorm.moving_mean
                conv_inner.batchnorm.moving_variance = subcell.batchnorm.moving_variance
            del subcell.batchnorm
            subcell.batchnorm = None
            subcell.has_bn = False
        else:
            conv_inner = quant.Conv2dQuant(conv_inner.in_channels, conv_inner.out_channels,
                                           kernel_size=conv_inner.kernel_size, stride=conv_inner.stride,
                                           pad_mode=conv_inner.pad_mode, padding=conv_inner.padding,
                                           dilation=conv_inner.dilation, group=conv_inner.group,
                                           has_bias=conv_inner.has_bias, quant_config=self.quant_config,
                                           quant_dtype=self.weight_dtype)
        # change original network Conv2D OP parameters to quant network
        conv_inner.weight = subcell.conv.weight
        if subcell.conv.has_bias:
            conv_inner.bias = subcell.conv.bias
        subcell.conv = conv_inner
        if subcell.has_act and subcell.activation is not None:
            subcell.activation = self._convert_activation(subcell.activation)
        elif subcell.after_fake:
            subcell.has_act = True
            subcell.activation = _AddFakeQuantAfterSubCell(F.identity, quant_dtype=self.act_dtype,
                                                           quant_delay=self.act_qdelay, per_channel=self.act_channel,
                                                           symmetric=self.act_symmetric, narrow_range=self.act_range,
                                                           optimize_option=self.optimize_option)
        return subcell

    def _convert_dense(self, subcell):
        """
        convert dense cell to quant cell
        """
        min_init = -6
        max_init = 6
        if OptimizeOption.LEARNED_SCALE in self.optimize_option:
            subcell_weight_para = subcell.dense.weight.data.asnumpy()
            if subcell.has_bn:
                scale_factor = (subcell.batchnorm.gamma.data.asnumpy() /
                                np.sqrt(subcell.batchnorm.moving_variance.data.asnumpy() + self.eps))
                subcell_weight_para = subcell_weight_para * scale_factor.reshape(-1, 1, 1, 1)
            min_init, max_init = self._KL_init(subcell_weight_para, self.weight_dtype)
        self.quant_config = self.quant_config._replace(
            weight=self.quant_config.weight.partial_init(min_init=min_init, max_init=max_init))

        dense_inner = subcell.dense
        dense_inner = quant.DenseQuant(dense_inner.in_channels,
                                       dense_inner.out_channels,
                                       has_bias=dense_inner.has_bias,
                                       quant_config=self.quant_config,
                                       quant_dtype=self.weight_dtype)
        # change original network Dense OP parameters to quant network
        dense_inner.weight = subcell.dense.weight
        if subcell.dense.has_bias:
            dense_inner.bias = subcell.dense.bias
        subcell.dense = dense_inner
        if subcell.has_act and subcell.activation is not None:
            subcell.activation = self._convert_activation(subcell.activation)
        elif subcell.after_fake:
            subcell.has_act = True
            subcell.activation = _AddFakeQuantAfterSubCell(F.identity,
                                                           quant_dtype=self.act_dtype,
                                                           quant_delay=self.act_qdelay,
                                                           per_channel=self.act_channel,
                                                           symmetric=self.act_symmetric,
                                                           narrow_range=self.act_range,
                                                           optimize_option=self.optimize_option)
        return subcell

    def _convert_activation(self, activation):
        """
        convert activation cell to quant cell
        """
        act_class = activation.__class__
        act_list = [nn.ReLU, nn.ReLU6, nn.Sigmoid]
        act_list_with_fake_before = [nn.LeakyReLU, nn.HSigmoid, nn.HSwish]

        if act_class in act_list:
            return quant.ActQuant(activation=activation,
                                  quant_config=self.quant_config,
                                  quant_dtype=self.act_dtype)
        if act_class in act_list_with_fake_before:
            return quant.ActQuant(activation=activation,
                                  ema=True,
                                  fake_before=True,
                                  quant_config=self.quant_config,
                                  quant_dtype=self.act_dtype)
        raise ValueError("Unsupported activation in auto quant: ", act_class)

    def _KL_init(self, subcell_weight_para, weight_dtype):
        """
        Calculate the value of max_init and min_init with compute_KL_threshold.
        """
        if self.weight_channel:
            max_init = [compute_KL_threshold(weight_para_each, weight_dtype)
                        for weight_para_each in subcell_weight_para]
            min_init = [-x for x in max_init]
        else:
            max_init = [compute_KL_threshold(subcell_weight_para, weight_dtype)]
            min_init = [-x for x in max_init]
        return min_init, max_init

    def _set_mixed_bits(self, network, strategy):
        r"""
        Set network's quantization strategy, this function is currently only valid for `LEARNED_SCALE`
        optimize_option.

        Args:
            network (Cell): Input network.
            strategy (list): The quantization strategy for layers that need to be quantified (eg. [[8], [8],
                ..., [6], [4], [8]]), currently only the quant_dtype for weights of the dense layer and the
                convolution layer is supported.

        Returns:
            Cell, a network with mixed bit strategy configured.

        Raises:
            ValueError: If `OptimizeOption.LEARNED_SCALE` is not in `self.optimize_option`.
        """
        if OptimizeOption.LEARNED_SCALE not in self.optimize_option:
            raise ValueError("The `_set_mixed_bits` function is currently only valid for `LEARNED_SCALE` "
                             "optimize_option.")

        self.quantizable_idx = []
        pass_cell = None
        for i, cell_and_name in enumerate(network.cells_and_names()):
            cell = cell_and_name[1]
            if isinstance(cell, (nn.Conv2dBnAct, nn.DenseBnAct)) and cell is not pass_cell:
                self.quantizable_idx.append(i)

        assert len(self.quantizable_idx) == len(strategy)
        quantizable_layer_bit_dict = {idx: bit for idx, bit in zip(self.quantizable_idx, strategy)}
        type_map = {
            QuantDtype.INT2.num_bits: QuantDtype.INT2,
            QuantDtype.INT3.num_bits: QuantDtype.INT3,
            QuantDtype.INT4.num_bits: QuantDtype.INT4,
            QuantDtype.INT5.num_bits: QuantDtype.INT5,
            QuantDtype.INT6.num_bits: QuantDtype.INT6,
            QuantDtype.INT7.num_bits: QuantDtype.INT7,
            QuantDtype.INT8.num_bits: QuantDtype.INT8
        }
        for i, cell_and_name in enumerate(network.cells_and_names()):
            cell = cell_and_name[1]
            if i not in self.quantizable_idx:
                continue
            else:
                if isinstance(cell, (nn.Conv2dBnAct, nn.DenseBnAct)):
                    cell.weight_dtype = type_map[quantizable_layer_bit_dict[i][0]]
                    if isinstance(cell, nn.Conv2dBnAct):
                        subcell_weight_para = cell.conv.weight.data.asnumpy()
                        if hasattr(cell.conv, 'gamma'):
                            scale_factor = (cell.conv.gamma.data.asnumpy() /
                                            np.sqrt(cell.conv.moving_variance.data.asnumpy() + self.eps))
                            subcell_weight_para = subcell_weight_para * scale_factor.reshape(-1, 1, 1, 1)
                        min_init, max_init = self._KL_init(subcell_weight_para, cell.weight_dtype)
                        cell.conv.fake_quant_weight.reset(quant_dtype=cell.weight_dtype,
                                                          min_init=min_init,
                                                          max_init=max_init)
                    elif isinstance(cell, nn.DenseBnAct):
                        subcell_weight_para = cell.dense.weight.data.asnumpy()
                        if hasattr(cell.dense, 'gamma'):
                            scale_factor = (cell.dense.gamma.data.asnumpy() /
                                            np.sqrt(cell.dense.moving_variance.data.asnumpy() + self.eps))
                            subcell_weight_para = subcell_weight_para * scale_factor.reshape(-1, 1, 1, 1)
                        min_init, max_init = self._KL_init(subcell_weight_para, cell.weight_dtype)
                        cell.dense.fake_quant_weight.reset(quant_dtype=cell.weight_dtype,
                                                           min_init=min_init,
                                                           max_init=max_init)
        return network


text:
import random
import numpy as np
import sys
from PyQt5.QtWidgets import QWidget, QApplication, QLabel, QGridLayout, QMessageBox
from PyQt5.QtGui import QFont, QPalette
from PyQt5.QtCore import Qt

class Block(QLabel):
	def __init__(self, text):
		super().__init__()

		# Properties of the grids
		self.Text = text
		self.Size = 80
		self.Background = "orange"
		self.ZeroBackground = "gray"
		self.BorderRadius = "10px"

		#properties of text on the block
		self.Font = "Arial"
		self.FontSize = 30
		self.Bold = True
		self.Alignment = "center"
		self.initUI()

	def initUI(self):
		# Block
		self.setFixedSize(self.Size, self.Size)
		# Text
		font = QFont(self.Font, self.FontSize)
		font.setBold(self.Bold)
		self.setFont(font)

		# Set color of text
		pa = QPalette()
		pa.setColor(QPalette.WindowText, Qt.white)
		self.setPalette(pa)

		self.setStyleSheet("border-radius:" + self.BorderRadius)
		self.setAlignment(Qt.AlignCenter)

		# If number == 0, block background should be the same as window background
		if self.Text ==0:
			self.setStyleSheet("background-color:%s;" % self.ZeroBackground)
		else:
			self.setText(str(self.Text))
			self.setStyleSheet("background: %s;" % self.Background)

class Puzzle(QWidget):

	def __init__(self):
		# Basic stuff of the game
		super().__init__()
		self.Title = 'Puzzle Game'
		# Size of rows and columns
		self.PuzzleSize = 4
		self.background = 'gray'
		# create 2d array with random number from 0 to 15
		self.matrix = self.CreateMatrix()
		# layout of the window
		self.Top = 10
		self.Left = 10
		self.Size = 400

		self.ZeroRow, self.ZeroCol = self.ZeroPos()

		self.layout = QGridLayout()

		self.initUI()

	def initUI(self):
		self.setWindowTitle(self.Title)
		self.setGeometry(self.Left,self.Top, self.Size, self.Size)
		self.setStyleSheet("background:" + self.background)

		self.SetLayout()

	def CreateMatrix(self):

		# get an array with random int from 0 to 16
		blocks = list(range(0,16))
		# mess around the elements in array to make sure the number is random
		random.shuffle(blocks)
		# convert 1d array into 2d array
		matrix = [blocks[x:x+4] for x in range(0,len(blocks),4)]

		return matrix

	def ZeroPos(self):
		# find zero in the matrix
		for items in self.matrix: # go through every sub-list
			if 0 in items:
				zeroRow = self.matrix.index(items) #find out 0 in which sub list
				zeroCol = items.index(0)
		return zeroRow, zeroCol

	def SetLayout(self):

		for row in range(4):
			for col in range(4):
				self.layout.addWidget(Block(self.matrix[row][col]), row, col)
		self.setLayout(self.layout)
		self.show()

	def keyPressEvent(self, event):
		key = event.key()

		if (key == Qt.Key_Up or key == Qt.Key_W):
			self.move('up')
		elif (key == Qt.Key_Down or key == Qt.Key_S):
			self.move('down')
		elif (key == Qt.Key_Left or key == Qt.Key_A):
			self.move('left')
		elif (key == Qt.Key_Right or key == Qt.Key_D):
			self.move('right')

	def move(self, direction):
		ZeroRow = self.ZeroRow
		ZeroCol = self.ZeroCol
		if direction == 'up' and self.ZeroRow != 3:
			# swap position of 0 and the one below it
			self.matrix[ZeroRow][ZeroCol], self.matrix[ZeroRow+1][ZeroCol] = self.matrix[ZeroRow+1][ZeroCol], self.matrix[ZeroRow][ZeroCol]
			# update position of zero
			self.ZeroRow += 1

		if direction == 'down' and self.ZeroRow != 0:
			# swap position of 0 and the one below it
			self.matrix[ZeroRow][ZeroCol], self.matrix[ZeroRow-1][ZeroCol] = self.matrix[ZeroRow-1][ZeroCol], self.matrix[ZeroRow][ZeroCol]
			# update position of zero
			self.ZeroRow -= 1

		if direction == 'left' and self.ZeroCol != 3:
			# swap position of 0 and the one below it
			self.matrix[ZeroRow][ZeroCol], self.matrix[ZeroRow][ZeroCol+1] = self.matrix[ZeroRow][ZeroCol+1], self.matrix[ZeroRow][ZeroCol]
			# update position of zero
			self.ZeroCol+= 1

		if direction == 'right' and self.ZeroCol != 0:
			# swap position of 0 and the one below it
			self.matrix[ZeroRow][ZeroCol], self.matrix[ZeroRow][ZeroCol-1] = self.matrix[ZeroRow][ZeroCol-1], self.matrix[ZeroRow][ZeroCol]
			# update position of zero
			self.ZeroCol -= 1

		self.SetLayout()
		# After each move, check if use find the right solution
		# if they do, message box bumps up
		# click ok to restart another game
		if (self.CheckResult()):
			# message box bumps up
			# user click ok to restart the game
			print(self.matrix)
			if QMessageBox.Ok == QMessageBox.information(self, 'Congradulations!', 'You win!!!'):
				self.matrix = self.CreateMatrix()
				self.SetLayout()

	def CheckResult(self):
		# first check if 0 is the last
		if self.matrix[self.PuzzleSize-1][self.PuzzleSize-1] != 0:
			return False
		# check numbers are in right position
		for row in range(self.PuzzleSize):
			for col in range(self.PuzzleSize):
				if self.matrix[row][col] != row * self.PuzzleSize + col + 1:
					# user find the solution
					if row == 3 and col == 3:
						pass
					else:
						print(row,col)
						return False

		return True

if __name__ == '__main__':
	app = QApplication(sys.argv)
	ex = Puzzle()
	sys.exit(app.exec_())

ext: py

reason: import numpy


text:
import torch
from torch import nn
import numpy as np
from torch.nn import functional as F

class BaseAttentionNetwork(nn.Module):
    def __init__(self, sentence_encoder, relation_levels, relation_level_layer, keep_prob,
                 train_batch_size=None, test_batch_size=262, num_classes=53, device = "cuda:0"):
        '''
        Pay Attention!
        relation_matrix:
        id = 0: virtual node
        id = 1: NA node
        '''
        super(BaseAttentionNetwork, self).__init__()
        self.keep_prob = keep_prob
        self.hidden_size = sentence_encoder.hidden_size
        self.relation_levels = relation_levels
        self.sentence_encoder = sentence_encoder

        self.train_batch_size = train_batch_size
        self.test_batch_size = test_batch_size

        self.hier = relation_levels.shape[1]
        self.layer = relation_level_layer
        self.num_classes = num_classes
        self.relation_matrixs = []
        self.device = device

        self.discrimitive_matrix = nn.Parameter(torch.Tensor(num_classes, self.hidden_size * 2))
        self.bias = torch.nn.Parameter(torch.Tensor(53))
        self.drop = nn.Dropout(1 - self.keep_prob)
        self.long_tail = []
        self.normal_body = []
        self.short_head = []

    def reset_parameters(self):
        NotImplemented
    def forward(self):
        NotImplemented
    def forward_infer(self):
        NotImplemented
    def query_func(self, attention_weight, label):
        NotImplemented

class HAttentionNetwork(BaseAttentionNetwork):
    def __init__(self, sentence_encoder, relation_levels, relation_level_layer, keep_prob,
                 train_batch_size=None, test_batch_size=262, num_classes=53, device = "cuda:0"):
        '''
        Pay Attention!
        relation_matrix:
        id = 0: virtual node
        id = 1: NA node
        '''
        super(HAttentionNetwork, self).__init__(sentence_encoder = sentence_encoder,
                                                   relation_levels = relation_levels,
                                                   relation_level_layer = relation_level_layer,
                                                   keep_prob = keep_prob,
                                                   train_batch_size=train_batch_size,
                                                   test_batch_size=test_batch_size,
                                                   num_classes=num_classes,
                                                   device = device)
        self.reset_parameters()

    def reset_parameters(self):
        for i in range(self.hier):
            self.relation_matrixs.append(nn.Embedding(self.layer[i], self.hidden_size, _weight=nn.init.xavier_uniform_(
                torch.Tensor(self.layer[i], self.hidden_size))).to(self.device))
        nn.init.xavier_uniform_(self.discrimitive_matrix)
        nn.init.zeros_(self.bias)

    def forward(self, data):  # data 包含word, pos1, pos2, mask, label, scope
        x = self.sentence_encoder(data)

        label_layer = self.relation_levels[data['label_index']]
        attention_logits = []
        for i in range(self.hier):
            current_relation = self.relation_matrixs[i](label_layer[:, i])  # batch * 230
            attention_logits.append(torch.sum(current_relation * x, 1))  # batch*230 x batch*230
        #             print("torch.sum(current_relation * x, 1):" ,torch.sum(current_relation * x, 1).shape)

        attention_logits_stack = torch.stack(attention_logits)  # 将一个batch的结果堆叠起来   2 * batch_size
        #         print("attention_logits stack shape:", attention_logits_stack.shape)

        attention_score_hidden = torch.cat([
            F.softmax(attention_logits_stack[:, data['scope'][i]:data['scope'][i + 1]], dim = -1) for i in
            range(self.train_batch_size)], 1)  ###这段出了问题

        tower_repre = []
        for i in range(self.train_batch_size):
            sen_matrix = x[data['scope'][i]:data['scope'][i + 1]]
            #             print("sen_matrix shape: ",sen_matrix.shape)
            layer_score = attention_score_hidden[:,
                          data['scope'][i]:data['scope'][i + 1]]  # 查找Layer_score #(2 ,bag_size)
            #             print("layer_score shape: ",layer_score.shape)
            layer_repre = torch.reshape(layer_score @ sen_matrix, [-1])  # 获得层次化表示表示  # 2 * batch_size @ batch_size *230
            #             print("layer_score @ sen_matrix shape: ", (layer_score @ sen_matrix).shape) #(2, 230)
            tower_repre.append(layer_repre)  # 获得每个句子的表示  append (-1, 460)
        #             print("layer_repre shape: ", (layer_repre).shape)

        stack_repre = self.drop(torch.stack(tower_repre))  # 获得新的表示
        logits = stack_repre @ self.discrimitive_matrix.t() + self.bias  # sen_num * 230 matmul 230 * 53 + 53
        return logits

    def forward_infer(self, data):
        x = self.sentence_encoder(data)  # batch_size * 230

        test_attention_scores = []
        for i in range(self.hier):
            current_relation = self.relation_matrixs[i](self.relation_levels[:, i])  # 53 * 230
            current_logit = current_relation @ x.t()  # 53 * batch_size
            current_score = torch.cat([F.softmax(current_logit[:, data['scope'][j]:data['scope'][j + 1]], dim = -1) for j in
                                       range(self.test_batch_size)], 1)  ##得到每一个袋子的attention_score
            test_attention_scores.append(
                current_score)  # curr_relation_num * batch_size   堆叠起来，形成两个高度不同的attention_score: 53 * batch_size
        #             print(current_relation)
        #             print("torch.sum(current_relation * x, 1):" ,torch.sum(current_relation * x, 1).shape)

        test_attention_scores_stack = torch.stack(test_attention_scores, 1)  # 将一个batch的结果堆叠起来: 53* 2* batch_size
        #         print("attention_logits stack shape:", attention_logits_stack.shape)

        test_tower_output = []
        for i in range(self.test_batch_size):
            test_sen_matrix = (torch.unsqueeze(x[data['scope'][i]:data['scope'][i + 1]], 0)).repeat(53, 1,
                                                                                1)  # 先将 x[data['scope']] 扩充维度形成  53 * bag_size * 230， 然后在第一维重复53次
            #             print("sen_matrix shape: ",sen_matrix.shape)
            test_layer_score = test_attention_scores_stack[:, :,
                               data['scope'][i]:data['scope'][i + 1]]  # 查找Layer_score #(53, 2 ,bag_size)
            #             print("layer_score shape: ",layer_score.shape)
            test_layer_repre = torch.reshape(test_layer_score @ test_sen_matrix, [self.num_classes,
                                                                                -1])  # 获得层次化表示表示  # 53 * 2 * bag_size @ 53* bag_size *230, 53* 2 *230 ,reshape 53 *460
            #             print("layer_score @ sen_matrix shape: ", (layer_score @ sen_matrix).shape) #(2, 230) (r_(h, t)^1), (r_(h, t)^2)
            #             print("layer_repre shape: ", (layer_repre).shape)
            test_logits = test_layer_repre @ self.discrimitive_matrix.t() + self.bias  # 53 * 460 @ 460 * 53 +   (53, )
            test_output = torch.diagonal(F.softmax(test_logits, dim = -1))
            test_tower_output.append(test_output)

        test_stack_output = torch.reshape(torch.stack(test_tower_output), [self.test_batch_size, self.num_classes])
        return test_stack_output

class KAttentionNetwork(BaseAttentionNetwork):
    def __init__(self, sentence_encoder,  relation_levels, relation_level_layer, attention_weight, relation_matrixs, keep_prob,
                 train_batch_size=262, test_batch_size=None, num_classes=53, device = "cuda:0"):
        '''
        Pay Attention!
        relation_matrix:
        id = 0: virtual node
        id = 1: NA node
        '''
        super(KAttentionNetwork, self).__init__(sentence_encoder = sentence_encoder,
                                                   relation_levels = relation_levels,
                                                   relation_level_layer = relation_level_layer,
                                                   keep_prob = keep_prob,
                                                   train_batch_size=train_batch_size,
                                                   test_batch_size=test_batch_size,
                                                   num_classes=num_classes,
                                                   device = device)

        self.relation_matrixs = []
        self.attention_weight = attention_weight

        self.w_s = nn.Parameter(torch.Tensor(self.hier, self.hidden_size * 3))
        self.b_s = nn.Parameter(torch.Tensor(self.hier, 1))

        self.reset_parameters(relation_matrixs)

    def reset_parameters(self, relation_matrixs):
        for i in range(self.hier):
            self.relation_matrixs.append(nn.Embedding(self.layer[i], self.hidden_size, _weight=relation_matrixs[i]).to(self.device))
        nn.init.xavier_uniform_(self.discrimitive_matrix)
        nn.init.zeros_(self.bias)

    def forward(self, data):  # data 包含word, pos1, pos2, mask, label, scope
        x = self.sentence_encoder(data)
        label_layer = self.relation_levels[data['label_index']]
        s_k_q_r = x #total_size * 230
        vertical_attention_logits = []
        for i in range(self.hier):
            current_relation = self.relation_matrixs[i](label_layer[:, i])  # total_size * 230
            s_k_q_r = torch.cat((s_k_q_r, current_relation), dim=-1)  # total_size * 690
            current_ver_attention = self.attention_weight[i][label_layer[:, i]]
            vertical_attention_logits.append(current_ver_attention) # [total_size, total_size]

        attention_logits_stack = self.w_s @ torch.tanh(s_k_q_r).t() + self.b_s #  2* 690  @ 690 * total_sie = 2 * total_size
        attention_score_hidden = torch.cat([
            F.softmax(attention_logits_stack[:, data['scope'][i]:data['scope'][i + 1]], dim = -1) for i in
            range(self.train_batch_size)], 1)  ###这段出了问题 #

        vertical_attention_logits_stack = torch.stack(vertical_attention_logits)  # 2* toltal_size

        tower_repre = []
        for i in range(self.train_batch_size):
            sen_matrix = x[data['scope'][i]:data['scope'][i + 1]]
            #             print("sen_matrix shape: ",sen_matrix.shape)
            layer_score = attention_score_hidden[:,
                          data['scope'][i]:data['scope'][i + 1]]  # 查找Layer_score #(2 ,bag_size)
            layer_attention_logits = vertical_attention_logits_stack[:, data['scope'][i]:data['scope'][i + 1]].permute( 1, 0)  # (bag_size, 2)
            layer_attention_score = torch.mean(F.softmax(layer_attention_logits, dim=-1).permute(1, 0), dim=1).reshape(2,1) #(2 * 1)
            layer_repre = torch.reshape( layer_attention_score * (layer_score @ sen_matrix), [-1])  # 获得层次化表示表示  # 2* 690 = 2*bag_size @ bag_size* 690
            #             print("layer_score @ sen_matrix shape: ", (layer_score @ sen_matrix).shape) #(2, 230) (r_(h, t)^1), (r_(h, t)^2)
            tower_repre.append(layer_repre)  # 获得每个句子的表示 (batchsize, bagsize*230 )
        #             print("layer_repre shape: ", (layer_repre).shape)

        stack_repre = self.drop(torch.stack(tower_repre))  # 获得新的表示
        logits = stack_repre @ self.discrimitive_matrix.t() + self.bias  # sen_num * 230 matmul 230 * 53 + 53
        return logits

    def forward_infer(self, data):
        x = self.sentence_encoder(data)  # total_size * 230

        test_attention_scores = []
        s_k_q_r = x #total_size *30
        s_k_q_r = torch.unsqueeze(s_k_q_r, dim = 0).repeat(self.num_classes, 1, 1)
        test_vertical_attention_logits = []
        for i in range(self.hier):
            current_relation = self.relation_matrixs[i](self.relation_levels[:, i])  # 53 * 230
            current_relation = torch.unsqueeze(current_relation, 1).repeat(1, x.shape[0], 1) # 53 * total_size* 230
            s_k_q_r = torch.cat((s_k_q_r, current_relation), dim = -1) # 53 * total_size * 690

            test_current_ver_attention = self.attention_weight[i][self.relation_levels[:, i]] # 53 *1
            test_vertical_attention_logits.append(test_current_ver_attention) # [53, 53]

        test_attention_score_stack = self.w_s @ torch.tanh(s_k_q_r).permute(0, 2, 1) + self.b_s # 53 * 2 * total_size

        vertical_attention_logits_stack = torch.stack(test_vertical_attention_logits)  # 2* 53
        test_layer_attention_score = torch.unsqueeze(F.softmax(vertical_attention_logits_stack.t()), dim = -1) # 53* 2 * 1
        test_tower_output = []
        for i in range(self.test_batch_size):
            test_sen_matrix = (torch.unsqueeze(x[data['scope'][i]:data['scope'][i + 1]], 0)).repeat(53, 1,
                                                                                1)  # 先将 x[data['scope']] 扩充维度形成  53 * bag_size * 230， 然后在第一维重复53次
            test_layer_score = test_attention_score_stack[:, :,
                               data['scope'][i]:data['scope'][i + 1]]  # 查找Layer_score #(53, 2 ,bag_size)

            test_layer_repre = torch.reshape(test_layer_attention_score * (test_layer_score @ test_sen_matrix), [self.num_classes,
                                                                                -1])  # 获得层次化表示表示  # 53 * 2 * bag_size @ 53* bag_size *230, 53* 2 *230 ,reshape 53 *460

            test_logits = test_layer_repre @ self.discrimitive_matrix.t() + self.bias  # 53 * 460 @ 460 * 53 +   (53, )
            test_output = torch.diagonal(F.softmax(test_logits))  # 获得对角输出
            test_tower_output.append(test_output)

        test_stack_output = torch.reshape(torch.stack(test_tower_output), [self.test_batch_size, self.num_classes])
        return test_stack_output

max_stars_repo_name: JohannesLiu/HNRE-Pytorch

max_issues_repo_name: JohannesLiu/HNRE-Pytorch

max_forks_repo_name: JohannesLiu/HNRE-Pytorch


text:
from pyristic.utils.helpers import function_type
from tqdm import tqdm
import numpy as np

__all__= ['Genetic']

class Genetic:

    def __init__(self,  function: function_type,\
                        decision_variables:int,\
                        constraints:list=[],\
                        bounds: list=[],\
                        config = None):
        """
        ------------------------------------------------------
        Description:
            Initializing every variable necessary to the search.
        Arguments:
            - Function: Objective function to minimize.
            - Constraints: Constraints to be a feasible solution.
            - Bounds: bound for every variable, this should be a matrix 2 x N
            where N is the variables number. The first element is lowert limit
            and another one is the upper limit.
        ------------------------------------------------------
        """
        #Information about problem.
        self.f = function
        self.Constraints = constraints
        self.Bounds = bounds
        self.Decision_variables = decision_variables # Decision variables.

        #Operators.
        if config != None:
            self._mutation_operator  = config.mutation_op
            self._crossover_operator = config.cross_op
            self._survivor_selector  = config.survivor_selector
            self._parent_selector    = config.parent_selector
            self._fixer              = config.fixer
        #Search information.
        self.logger = {}
        self.logger['best_individual']      = None
        self.logger['best_f']        = None
        self.logger['current_iter']    = None
        self.logger['total_iter']      = None
        self.logger['population_size'] = None

    def __str__(self):
        printable = "Genetic search: \n F_a(X) = {} \n X = {} \n ".format(self.logger['best_f'], self.logger['best_individual'])
        first = True

        for i in range(len(self.Constraints)):
            if self.Constraints[i].__doc__ != None:

                if first:
                    first = False
                    printable += "Constraints: \n "

                self.Constraints[i](self.logger['best_individual'])
                printable += "{} \n".format( self.Constraints[i].__doc__)

        return printable

    #-----------------------------------------------------
                    #Public functions.
    #-----------------------------------------------------
    def optimize(self,  generations:int ,\
                        size_population: int,\
                        cross_percentage: float = 1.0,\
                        verbose:bool=True,\
                        **kwargs) -> None:
        """
        ------------------------------------------------------
        Description:
            The main function to find the best solution using tabu search.
        Arguments:
            -generations: number of iterations.
            -population: Number of new solutions.
        ------------------------------------------------------
        """
        assert 0 <= cross_percentage <= 1.0
        #Reset global solution.
        self.logger['current_iter']    = 0
        self.logger['total_iter']      = generations
        self.logger['population_size'] = size_population
        self.logger['best_f']        = None
        self.logger['best_individual']      = None

        #Initial population.
        self.logger['parent_population_x']       = self.initialize_population(**kwargs)
        self.logger['parent_population_f'] = np.apply_along_axis(self.f , 1, self.logger['parent_population_x'])

        try:
            for g in tqdm(range(generations), disable = not verbose):
                #Parent selection.
                parent_ind = self.parent_selection(**kwargs)
                first_parent_indices, second_parent_indices = self.get_pairs(parent_ind)
                crossLenght = int(len(first_parent_indices) * cross_percentage)
                first_parent_cross, second_parent_cross =first_parent_indices[:crossLenght], second_parent_indices[:crossLenght]
                first_parent_without_cross, second_parent_without_cross = first_parent_indices[crossLenght:], second_parent_indices[crossLenght:]
                #Crossover.
                self.logger['offspring_population_x'] = self.crossover_operator(first_parent_cross,\
                                                                                second_parent_cross,\
                                                                                **kwargs)
                if len(first_parent_without_cross) != 0:
                    parent_indices = np.concatenate((first_parent_without_cross,second_parent_without_cross))
                    parents = self.logger['parent_population_x'][parent_indices]
                    self.logger['offspring_population_x'] = np.concatenate((self.logger['offspring_population_x'], parents), axis = 0)

                #mutate.
                self.logger['offspring_population_x'] = self.mutation_operator(**kwargs)

                #Fixing solutions and getting aptitude.
                f_offspring = []
                for ind in range(len(self.logger['offspring_population_x'])):
                    if self.is_invalid(self.logger['offspring_population_x'][ind]):
                        self.logger['offspring_population_x'][ind] = self.fixer(ind)
                    f_offspring.append(self.f(self.logger['offspring_population_x'][ind]))
                self.logger['offspring_population_f'] = np.array(f_offspring)

                #Survivor selection.
                next_generation = self.survivor_selection(**kwargs)
                self.logger['parent_population_x']  = next_generation['population']
                self.logger['parent_population_f']  = next_generation['parent_population_f']

                self.logger['current_iter'] += 1

        except KeyboardInterrupt:
            print("Interrupted, saving best solution found so far.")

        ind = np.argmin(self.logger['parent_population_f'])
        self.logger['best_individual'] = self.logger['parent_population_x'][ind]
        self.logger['best_f']   = self.logger['parent_population_f'][ind]

    def initialize_population(self, **kwargs) -> np.ndarray:
        """
        ------------------------------------------------------
        Description:
            How to distribute the population. This function create a
            population using uniform distribution.
            This should return an matrix of size (M x N)
            where M is the number of population and N is the number of
            variables.
        Arguments:
            -size_: tuple with two integers (m,n)
            where m is the number generated of new solutions and
            n is the number of variables about the problem.
        ------------------------------------------------------
        """
        return np.random.uniform( self.Bounds[0], self.Bounds[1], size=(self.logger['population_size'], self.Decision_variables))

    def fixer(self, ind: int) -> np.ndarray:
        """
        ------------------------------------------------------
        Description:
        Function which helps to move solution in a valid region.
        Arguments:
            -X: numpy array, this one is the current population.
            -ind: index of individual.
        ------------------------------------------------------
        """
        if self._fixer == None:
            raise NotImplementedError
        return self._fixer(self.logger['offspring_population_x'], ind)

    def mutation_operator(self, **kwargs):
        """
        ------------------------------------------------------
        Description:
            The current population is updated by specific change
            using step_size.
            This function should change the population and current step size
            for the next generation.
        Arguments:
            -Population: Matrix with size m x n, where m is the current size of
            population and n is the problem variables. Every row is an element.
            - step_size: Matrix with size m x n, where m is the current size of
            population and n is the poblem variables. Every row is an step for
            every element.
        ------------------------------------------------------
        """
        if self._mutation_operator == None:
            raise NotImplementedError
        return self._mutation_operator(self.logger['offspring_population_x'])

    def crossover_operator(self,parent_ind1: np.ndarray,\
                                parent_ind2: np.ndarray,\
                                **kwargs):
        if self._crossover_operator == None:
            raise NotImplementedError
        return self._crossover_operator(self.logger['parent_population_x'], parent_ind1, parent_ind2)

    def survivor_selection(self,**kwargs) -> dict:
        if self._survivor_selector == None:
            raise NotImplementedError

        individuals = {}
        individuals['population'] = [self.logger['parent_population_x'], self.logger['offspring_population_x']]

        return self._survivor_selector( self.logger['parent_population_f'],\
                                        self.logger['offspring_population_f'],\
                                        individuals)

    def parent_selection(self,**kwargs) -> np.ndarray:
        if self._parent_selector == None:
            raise NotImplementedError
        return self._parent_selector(self.logger['parent_population_f'])

    #-----------------------------------------------------
                    #Private functions.
    #-----------------------------------------------------
    def is_invalid(self, x : np.ndarray) -> bool:
        """
        ------------------------------------------------------
        Description:
            Check if the current solution is invalid.
        ------------------------------------------------------
        """
        for constraint in self.Constraints:
            if not constraint(x):
                return True
        return False

    def get_pairs(self,parent_ind : np.ndarray):
        parent_ind1=[]
        parent_ind2=[]
        size_ = len(parent_ind)//2

        for a,b in np.random.randint(0,len(parent_ind),size=(size_,2)):
            parent_ind1.append(parent_ind[a])
            parent_ind2.append(parent_ind[b])

        return parent_ind1,parent_ind2


text:
import numpy as np
from pp.geo_utils import path_length
from pp.routing.manhattan import remove_flat_angles
from pp.routing.manhattan import _is_horizontal
from pp.routing.manhattan import _is_vertical

def path_length_matched_points(
    list_of_waypoints, modify_segment_i=-2, bend_radius=10.0, dL0=0.0, nb_loops=1
):
    """
    Several types of paths won't match correctly.
    We do not try to handle all the corner cases here.
    If the paths are not well behaved enough, the input list_of_waypoints needs to be modified.

    Args:
        list_of_waypoints:  [[p1, p2, p3,...], [q1, q2, q3,...], ...] the number of turns have to be identical (usually means same number of points. exception is if there are some flat angles)
        bend_radius: used to estimate the position of new waypoints to accommodate bends with a given radius
        margin: some extra space to budget for in addition to the bend radius in most cases, the default is fine
        dL0: distance added to all path length compensation.  Useful is we want to add space for extra taper on all branches
        modify_segment_i: index of the segment which accomodates the new turns default is next to last segment

        nb_loops: number of extra loops added in the path
            if nb_loops==0, no extra loop is added, instead, in each route,
            the segment indexed by `modify_segment_i` is elongated to match
            the longuest route in `list_of_waypoints`

    returns: another list of waypoints where
        - the path_lenth of each waypoints list are identical
        - the number of turns are identical

    """

    common_params = {
        "list_of_waypoints": list_of_waypoints,
        "modify_segment_i": modify_segment_i,
        "dL0": dL0,
    }

    if nb_loops >= 1:
        return path_length_matched_points_add_waypoints(
            **common_params, bend_radius=bend_radius, nb_loops=nb_loops
        )
    else:
        return path_length_matched_points_modify_segment(**common_params)

def path_length_matched_points_modify_segment(list_of_waypoints, modify_segment_i, dL0):
    if type(list_of_waypoints) != list:
        raise ValueError(
            "list_of_waypoints should be a list, got {}".format(type(list_of_waypoints))
        )
    list_of_waypoints = [
        remove_flat_angles(waypoints) for waypoints in list_of_waypoints
    ]
    lengths = [path_length(waypoints) for waypoints in list_of_waypoints]
    L0 = max(lengths)

    N = len(list_of_waypoints[0])

    # Find how many turns there are per path
    nb_turns = [len(waypoints) - 2 for waypoints in list_of_waypoints]

    # The paths have to have the same number of turns, otherwise this algo
    # cannot path length match
    if min(nb_turns) != max(nb_turns):
        raise ValueError(
            "Number of turns in paths have to be identical got \
        {}".format(
                nb_turns
            )
        )

    if modify_segment_i < 0:
        modify_segment_i = modify_segment_i + N + 1

    list_new_waypoints = []

    # For each list of waypoints, modify one segment in-place
    for i, waypoints in enumerate(list_of_waypoints):
        p_s0, p_s1, p_next = waypoints[modify_segment_i - 1 : modify_segment_i + 2]

        p_s0 = np.array(p_s0)
        p_s1 = np.array(p_s1)

        L = lengths[i]

        # Path length compensation length
        dL = (L0 - L) / 2

        # Additional fixed length
        dL = dL + dL0

        # Modify the segment to accomodate for path length matching
        # Two cases: vertical or horizontal segment
        if _is_vertical(p_s0, p_s1):
            sx = np.sign(p_next[0] - p_s1[0])

            dx = -sx * dL
            dp = (dx, 0)
            # Sequence of displacements to apply

        elif _is_horizontal(p_s0, p_s1):
            sy = np.sign(p_next[1] - p_s1[1])

            dy = -sy * dL
            dp = (0, dy)

        waypoints[modify_segment_i - 1] = p_s0 + dp
        waypoints[modify_segment_i] = p_s1 + dp

        list_new_waypoints += [waypoints]
    return list_new_waypoints

def path_length_matched_points_add_waypoints(
    list_of_waypoints,
    modify_segment_i=-2,
    bend_radius=10.0,
    margin=0.5,
    dL0=0.0,
    nb_loops=1,
):

    """
    Args:
        list_of_waypoints: a list of list_of_points:
            [[p1, p2, p3,...], [q1, q2, q3,...], ...]
            - the number of turns have to be identical
                (usually means same number of points. exception is if there are
                some flat angles)

        bend_radius: used to estimate the position of new waypoints to accommodate
                    bends with a given radius

        margin: some extra space to budget for in addition to the bend radius
            in most cases, the default is fine

        dL0: distance added to all path length compensation.
            Useful is we want to add space for extra taper on all branches

        modify_segment_i: index of the segment which accomodates the new turns
            default is next to last segment

        nb_loops: number of extra loops added in the path

    returns:
        another list of waypoints where:
            - the path_lenth of each waypoints list are identical
            - the number of turns are identical

    Several types of paths won't match correctly. We do not try to handle
    all the corner cases here. If the paths are not well behaved enough,
    the input list_of_waypoints needs to be modified.

    """

    print(list_of_waypoints[0])

    if type(list_of_waypoints) != list:
        raise ValueError(
            "list_of_waypoints should be a list, got {}".format(type(list_of_waypoints))
        )
    list_of_waypoints = [
        remove_flat_angles(waypoints) for waypoints in list_of_waypoints
    ]
    lengths = [path_length(waypoints) for waypoints in list_of_waypoints]
    L0 = max(lengths)

    N = len(list_of_waypoints[0])

    # Find how many turns there are per path
    nb_turns = [len(waypoints) - 2 for waypoints in list_of_waypoints]

    # The paths have to have the same number of turns, otherwise cannot path-length
    # match with this algorithm
    if min(nb_turns) != max(nb_turns):
        raise ValueError(
            "Number of turns in paths have to be identical got \
        {}".format(
                nb_turns
            )
        )

    # To have flexibility in the path length, we need to add 4 bends
    """
    One path has to be converted in this way:

                      ----
                      |  |
                      |  |  This length is adjusted to make all path with the same length
                      |  |
    --------  ===> ---|  |---
    """

    # Get the points for the segment we need to modify
    a = margin + bend_radius
    if modify_segment_i < 0:
        modify_segment_i = modify_segment_i + N + 1
    list_new_waypoints = []

    for i, waypoints in enumerate(list_of_waypoints):
        p_s0, p_s1, p_next = waypoints[modify_segment_i - 2 : modify_segment_i + 1]

        p_s1 = np.array(p_s1)

        L = lengths[i]

        # Path length compensation length
        dL = (L0 - L) / (2 * nb_loops)

        # Additional fixed length
        dL = dL + dL0

        # Generate a new sequence of points which will replace this segment
        # Two cases: vertical or horizontal segment
        if _is_vertical(p_s0, p_s1):
            sx = np.sign(p_next[0] - p_s1[0])
            sy = np.sign(p_s1[1] - p_s0[1])

            dx = sx * (2 * a + dL)
            dy = sy * 2 * a

            # First new point to insert
            q0 = p_s1 + (0, -2 * nb_loops * dy)

            # Sequence of displacements to apply
            seq = [(dx, 0), (0, dy), (-dx, 0), (0, dy)] * nb_loops
            seq.pop()  # Remove last point to avoid flat angle with next point

        elif _is_horizontal(p_s0, p_s1):
            sy = np.sign(p_next[1] - p_s1[1])
            sx = np.sign(p_s1[0] - p_s0[0])

            dx = sx * 2 * a
            dy = sy * (2 * a + dL)

            q0 = p_s1 + (-2 * dx * nb_loops, 0)
            # Sequence of displacements to apply
            seq = [(0, dy), (dx, 0), (0, -dy), (dx, 0)] * nb_loops
            seq.pop()  # Remove last point to avoid flat angle with next point

        # Generate points to insert
        qs = [q0]
        for dp in seq:
            qs += [qs[-1] + dp]

        # print()
        # print(nb_loops)
        # for q in qs:
        # print(q)
        # print()
        inserted_points = np.stack(qs, axis=0)
        waypoints = np.array(waypoints)

        # Insert the points
        new_points = np.vstack(
            [
                waypoints[: modify_segment_i - 1],
                inserted_points,
                waypoints[modify_segment_i - 1 :],
            ]
        )
        list_new_waypoints += [new_points]

    return list_new_waypoints

max_stars_repo_name: smartalecH/gdsfactory

max_issues_repo_name: smartalecH/gdsfactory

max_forks_repo_name: smartalecH/gdsfactory


text:
import numpy as np
import torch
from torch.utils.data import Dataset

class WDCDataset(Dataset):
    """
    PyTorch Dataset for the Semantic Type Detection model. To be used by a PyTorch DataLoader to feed batches to the model.
    """

    def __init__(self, features, labels, transform=None):
        """
        :param features: data that is used to be trained or tested
        :param labels: labels that are predicted
        """

        self.features = features
        self.labels = labels

        self.transform = transform

    def __getitem__(self, index):
        batch_sample = self.features[0][index], self.features[1][index], self.features[2][index], \
                       self.features[3][index], self.labels[index]  # 960, 201, 400, 27

        if self.transform:
            batch_sample = self.transform(batch_sample)

        return batch_sample

    def __len__(self):
        return len(self.features[0])

class ToTensor:
    # Convert ndarrays to Tensors
    def __call__(self, batch_sample):
        x, y, z, w, labels = batch_sample
        return torch.from_numpy(x).float(), torch.from_numpy(y).float(), torch.from_numpy(z).float(), \
               torch.from_numpy(w).float(), torch.from_numpy(np.asarray(labels))

max_stars_repo_name: shashankboosi/Mycroft

max_issues_repo_name: shashankboosi/Mycroft

max_forks_repo_name: shashankboosi/Mycroft


text:
#!/usr/bin/env python
# encoding: utf-8

# The MIT License (MIT)

# Copyright (c) 2016-2018 CNRS

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

# AUTHORS
# Hervé BREDIN - http://herve.niderb.fr

"""
Feature extraction

Usage:
  pyannote-speech-feature [--robust --parallel --database=<database.yml>] <experiment_dir> <database.task.protocol>
  pyannote-speech-feature check [--database=<database.yml>] <experiment_dir> <database.task.protocol>
  pyannote-speech-feature -h | --help
  pyannote-speech-feature --version

Options:
  <experiment_dir>           Set experiment root directory. This script expects
                             a configuration file called "config.yml" to live
                             in this directory. See "Configuration file"
                             section below for more details.
  <database.task.protocol>   Set evaluation protocol (e.g. "Etape.SpeakerDiarization.TV")
  --database=<database.yml>  Path to pyannote.database configuration file.
  --robust                   When provided, skip files for which feature extraction fails.
  --parallel                 When provided, process files in parallel.
  -h --help                  Show this screen.
  --version                  Show version.

Configuration file:
    The configuration of each experiment is described in a file called
    <experiment_dir>/config.yml, that describes the feature extraction process
    (e.g. MFCCs).

    ................... <experiment_dir>/config.yml ...................
    feature_extraction:
       name: YaafeMFCC
       params:
          e: False                   # this experiments relies
          De: True                   # on 11 MFCC coefficients
          DDe: True                  # with 1st and 2nd derivatives
          D: True                    # without energy, but with
          DD: True                   # energy derivatives
    ...................................................................

"""

import yaml
import os.path
import numpy as np
import functools
from docopt import docopt

from pyannote.database import FileFinder
from pyannote.database import get_unique_identifier
from pyannote.database import get_protocol

from pyannote.core.utils.helper import get_class_by_name

from pyannote.audio.features import Precomputed
from pyannote.audio.features.utils import get_audio_duration
from pyannote.audio.features.precomputed import PyannoteFeatureExtractionError

from multiprocessing import cpu_count, Pool

def init_feature_extraction(experiment_dir):

    # load configuration file
    config_yml = experiment_dir + '/config.yml'
    with open(config_yml, 'r') as fp:
        config = yaml.load(fp, Loader=yaml.SafeLoader)

    FeatureExtraction = get_class_by_name(
        config['feature_extraction']['name'],
        default_module_name='pyannote.audio.features')
    feature_extraction = FeatureExtraction(
        **config['feature_extraction'].get('params', {}))

    return feature_extraction

def process_current_file(current_file, file_finder=None, precomputed=None,
                         feature_extraction=None, robust=False):

    try:
        current_file['audio'] = file_finder(current_file)
    except ValueError as e:
        if not robust:
            raise PyannoteFeatureExtractionError(*e.args)
        return e

    uri = get_unique_identifier(current_file)
    path = precomputed.get_path(current_file)

    if os.path.exists(path):
        return

    try:
        features = feature_extraction(current_file)
    except PyannoteFeatureExtractionError as e:
        msg = 'Feature extraction failed for file "{uri}".'
        return msg.format(uri=uri)

    if features is None:
        msg = 'Feature extraction returned None for file "{uri}".'
        return msg.format(uri=uri)

    if np.any(np.isnan(features.data)):
        msg = 'Feature extraction returned NaNs for file "{uri}".'
        return msg.format(uri=uri)

    precomputed.dump(current_file, features)

    return

def helper_extract(current_file, file_finder=None, experiment_dir=None,
                   config_yml=None, feature_extraction=None,
                   robust=False):

    if feature_extraction is None:
        feature_extraction = init_feature_extraction(experiment_dir)

    precomputed = Precomputed(root_dir=experiment_dir)
    return process_current_file(current_file, file_finder=file_finder,
                                precomputed=precomputed,
                                feature_extraction=feature_extraction,
                                robust=robust)

def extract(protocol_name, file_finder, experiment_dir,
            robust=False, parallel=False):

    protocol = get_protocol(protocol_name, progress=False)

    # load configuration file
    config_yml = experiment_dir + '/config.yml'
    with open(config_yml, 'r') as fp:
        config = yaml.load(fp, Loader=yaml.SafeLoader)

    FeatureExtraction = get_class_by_name(
        config['feature_extraction']['name'],
        default_module_name='pyannote.audio.features')
    feature_extraction = FeatureExtraction(
        **config['feature_extraction'].get('params', {}))

    sliding_window = feature_extraction.sliding_window
    dimension = feature_extraction.dimension

    # create metadata file at root that contains
    # sliding window and dimension information

    precomputed = Precomputed(root_dir=experiment_dir,
                              sliding_window=sliding_window,
                              dimension=dimension)

    if parallel:

        extract_one = functools.partial(helper_extract,
                                        file_finder=file_finder,
                                        experiment_dir=experiment_dir,
                                        config_yml=config_yml,
                                        robust=robust)

        n_jobs = cpu_count()
        pool = Pool(n_jobs)
        imap = pool.imap

    else:

        feature_extraction = init_feature_extraction(experiment_dir)
        extract_one = functools.partial(helper_extract,
                                        file_finder=file_finder,
                                        experiment_dir=experiment_dir,
                                        feature_extraction=feature_extraction,
                                        robust=robust)
        imap = map

    for result in imap(extract_one, FileFinder.protocol_file_iter(
        protocol, extra_keys=['audio'])):
        if result is None:
            continue
        print(result)

def check(protocol_name, file_finder, experiment_dir):

    protocol = get_protocol(protocol_name)
    precomputed = Precomputed(experiment_dir)

    for subset in ['development', 'test', 'train']:

        try:
            file_generator = getattr(protocol, subset)()
            first_item = next(file_generator)
        except NotImplementedError as e:
            continue

        for current_file in getattr(protocol, subset)():

            try:
                audio = file_finder(current_file)
                current_file['audio'] = audio
            except ValueError as e:
                print(e)
                continue

            duration = get_audio_duration(current_file)

            try:
                features = precomputed(current_file)
            except PyannoteFeatureExtractionError as e:
                print(e)
                continue

            if not np.isclose(duration,
                              features.getExtent().duration,
                              atol=1.):
                uri = get_unique_identifier(current_file)
                print('Duration mismatch for "{uri}"'.format(uri=uri))

            if np.any(np.isnan(features.data)):
                uri = get_unique_identifier(current_file)
                print('NaN for "{uri}"'.format(uri=uri))

def main():

    arguments = docopt(__doc__, version='Feature extraction')

    db_yml = arguments['--database']
    file_finder = FileFinder(config_yml=db_yml)

    protocol_name = arguments['<database.task.protocol>']
    experiment_dir = arguments['<experiment_dir>']

    if arguments['check']:
        check(protocol_name, file_finder, experiment_dir)
    else:
        robust = arguments['--robust']
        parallel = arguments['--parallel']
        extract(protocol_name, file_finder, experiment_dir,
                robust=robust, parallel=parallel)

max_stars_repo_name: lgalmant/pyannote-audio

max_issues_repo_name: lgalmant/pyannote-audio

max_forks_repo_name: lgalmant/pyannote-audio


text:
import scannerpy
import cv2
import numpy as np
import utils as utils
from scannerpy import Database, DeviceType, Job, ColumnType, FrameType
from scannerpy.stdlib import pipelines

import subprocess
import os.path
import numpy as np

# A kernel file defines a standalone Python kernel which performs some computation by exporting a
# Kernel class.
@scannerpy.register_python_op()
class CalibrationClass(scannerpy.Kernel):
    # __init__ is called once at the creation of the pipeline. Any arguments passed to the kernel
    # are provided through a protobuf object that you manually deserialize. See resize.proto for the
    # protobuf definition.
    def __init__(self, config):
        self._w = config.args['w']
        self._h = config.args['h']
        self._A = config.args['A']
        self._R = config.args['R']
        self._T = config.args['T']

    # execute is the core computation routine maps inputs to outputs, e.g. here resizes an input
    # frame to a smaller output frame.
    def execute(self, frame: FrameType, mask: FrameType) -> FrameType:
        edge_sfactor = 0.5
        edges = utils.robust_edge_detection(cv2.resize(frame, None, fx=edge_sfactor, fy=edge_sfactor))
        edges = cv2.resize(edges, None, fx=1. / edge_sfactor, fy=1. / edge_sfactor)
        edges = cv2.Canny(edges.astype(np.uint8) * 255, 100, 200) / 255.0

        mask = cv2.dilate(mask[:, :, 0], np.ones((25, 25), dtype=np.uint8))

        edges = edges * (1 - mask)
        dist_transf = cv2.distanceTransform((1 - edges).astype(np.uint8), cv2.DIST_L2, 0)

        template, field_mask = utils.draw_field(self._A, self._R, self._T, self._h, self._w)

        II, JJ = (template > 0).nonzero()
        synth_field2d = np.array([[JJ, II]]).T[:, :, 0]

        field3d = utils.plane_points_to_3d(synth_field2d, self._A, self._R, self._T)

        self._A, self._R, self._T = utils.calibrate_camera_dist_transf(self._A, self._R, self._T, dist_transf, field3d)

        rgb = frame.copy()
        canvas, mask = utils.draw_field(self._A, self._R, self._T, self._h, self._w)
        canvas = cv2.dilate(canvas.astype(np.uint8), np.ones((15, 15), dtype=np.uint8)).astype(float)
        rgb = rgb * (1 - canvas)[:, :, None] + np.dstack((canvas * 255, np.zeros_like(canvas), np.zeros_like(canvas)))

        # result = np.dstack((template, template, template))*255

        out = rgb.astype(np.uint8)
        return out

movie_path = '/home/krematas/Mountpoints/grail/data/barcelona/test.mp4'
print('Detecting faces in movie {}'.format(movie_path))
movie_name = os.path.splitext(os.path.basename(movie_path))[0]

mask_path = '/home/krematas/Mountpoints/grail/data/barcelona/mask.mp4'
db = Database()
print('Ingesting video into Scanner ...')
input_tables, failed = db.ingest_videos([(movie_name, movie_path), ('mask', mask_path)], force=True)

print(db.summarize())
print('Failures:', failed)

cam_data = np.load('/home/krematas/Mountpoints/grail/data/barcelona/calib/00114.npy').item()

# db.register_op('Calibrate', [('frame', ColumnType.Video), ('mask', ColumnType.Video)], [('resized', ColumnType.Video)])

# Custom Python kernels for ops reside in a separate file, here calibrate_kernel.py.
# cwd = '/home/krematas/code/scanner/examples/apps/soccer'
# db.register_python_kernel('Calibrate', DeviceType.CPU, cwd + '/calibrate_kernel.py')

frame = db.sources.FrameColumn()
mask = db.sources.FrameColumn()

# Then we use our op just like in the other examples.
calibrate_video_class = db.ops.CalibrationClass(frame=frame, mask=mask, w=3840, h=2160, A=cam_data['A'], R=cam_data['R'], T=cam_data['T'])
output_op = db.sinks.FrameColumn(columns={'frame': calibrate_video_class})

job = Job(op_args={
    frame: input_tables[0].column('frame'),
    mask: input_tables[1].column('frame'),
    output_op: 'example_resized',
})
[out_table] = db.run(output_op, [job], force=True)
out_table.column('frame').save_mp4(movie_name + '_faces')

print('Successfully generated {:s}_faces.mp4'.format(movie_name))

max_stars_repo_name: apoms/scanner

max_issues_repo_name: apoms/scanner

max_forks_repo_name: apoms/scanner


text:
"""
The binary module enables plotting of binary isobaric phase diagrams.
"""
import numpy as np

from pycalphad import equilibrium
import pycalphad.variables as v
from pycalphad.plot.eqplot import eqplot

def binplot(dbf, comps, phases, conds, eq_kwargs=None, **plot_kwargs):
    """
    Calculate the binary isobaric phase diagram.
    This function is a convenience wrapper around equilibrium() and eqplot().

    Parameters
    ----------
    dbf : Database
        Thermodynamic database containing the relevant parameters.
    comps : list
        Names of components to consider in the calculation.
    phases : list
        Names of phases to consider in the calculation.
    conds : dict
        Maps StateVariables to values and/or iterables of values.
        For binplot only one changing composition and one potential coordinate each is supported.
    eq_kwargs : optional
        Keyword arguments to equilibrium().
    plot_kwargs : optional
        Keyword arguments to eqplot().

    Returns
    -------
    A phase diagram as a figure.

    Examples
    --------
    None yet.
    """
    eq_kwargs = eq_kwargs if eq_kwargs is not None else dict()
    indep_comp = [key for key, value in conds.items() if isinstance(key, v.Composition) and len(np.atleast_1d(value)) > 1]
    indep_pot = [key for key, value in conds.items() if ((key == v.T) or (key == v.P)) and len(np.atleast_1d(value)) > 1]
    if (len(indep_comp) != 1) or (len(indep_pot) != 1):
        raise ValueError('binplot() requires exactly one composition and one potential coordinate')
    indep_comp = indep_comp[0]
    indep_pot = indep_pot[0]

    full_eq = equilibrium(dbf, comps, phases, conds, **eq_kwargs)
    return eqplot(full_eq, x=indep_comp, y=indep_pot, **plot_kwargs)

max_stars_repo_name: jeffwdoak/pycalphad

max_issues_repo_name: abhinavsaboo/pycalphad

max_forks_repo_name: abhinavsaboo/pycalphad