How to use the sockeye.config.Config function in sockeye

#     http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file 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.

import tempfile
import os

import pytest

from sockeye import config


class ConfigTest(config.Config):
    yaml_tag = "!ConfigTest"

    def __init__(self, param, config=None):
        super().__init__()
        self.param = param
        self.config = config


def test_base_freeze():
    c = config.Config()
    c.param = 1
    assert c.param == 1
    c.freeze()
    with pytest.raises(AttributeError) as e:
        c.param = 2
    assert str(e.value) == "Cannot set 'param' in frozen config"

def get_encoder(config: 'EncoderConfig', prefix: str = '') -> 'Encoder':
    if isinstance(config, RecurrentEncoderConfig):
        return get_recurrent_encoder(config, prefix)
    elif isinstance(config, transformer.TransformerConfig):
        return get_transformer_encoder(config, prefix)
    elif isinstance(config, ConvolutionalEncoderConfig):
        return get_convolutional_encoder(config, prefix)
    elif isinstance(config, EmptyEncoderConfig):
        return EncoderSequence([EmptyEncoder(config)], config.dtype)
    elif isinstance(config, GraphConvolutionEncoderConfig):
        return get_gcn_encoder(config, prefix)
    else:
        raise ValueError("Unsupported encoder configuration")


class RecurrentEncoderConfig(config.Config):
    """
    Recurrent encoder configuration.

    :param rnn_config: RNN configuration.
    :param conv_config: Optional configuration for convolutional embedding.
    :param reverse_input: Reverse embedding sequence before feeding into RNN.
    :param dtype: Data type.
    """

    def __init__(self,
                 rnn_config: rnn.RNNConfig,
                 conv_config: Optional['ConvolutionalEmbeddingConfig'] = None,
                 reverse_input: bool = False,
                 dtype: str = C.DTYPE_FP32) -> None:
        super().__init__()
        self.rnn_config = rnn_config

# 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.

# List is needed for mypy, but not used in the code, only in special comments
from typing import Optional, List, Iterable  # NOQA pylint: disable=unused-import

import mxnet as mx

from sockeye.config import Config
from sockeye.layers import LayerNormalization, LHUC
from . import constants as C
from . import utils


class RNNConfig(Config):
    """
    RNN configuration.

    :param cell_type: RNN cell type.
    :param num_hidden: Number of RNN hidden units.
    :param num_layers: Number of RNN layers.
    :param dropout_inputs: Dropout probability on RNN inputs (Gal, 2015).
    :param dropout_states: Dropout probability on RNN states (Gal, 2015).
    :param dropout_recurrent: Dropout probability on cell update (Semeniuta, 2016).
    :param residual: Whether to add residual connections between multi-layered RNNs.
    :param first_residual_layer: First layer with a residual connection (1-based indexes).
           Default is to start at the second layer.
    :param forget_bias: Initial value of forget biases.
    :param lhuc: Apply LHUC (Vilar 2018) to the hidden units of the RNN.
    :param dtype: Data type.
    """

validation_target=validation_target,
                                               val_src_graphs=val_source_graphs,
                                               buckets=buckets,
                                               bucket_batch_sizes=bucket_batch_sizes,
                                               source_vocabs=source_vocabs,
                                               target_vocab=target_vocab,
                                               edge_vocab=edge_vocab,
                                               max_seq_len_source=max_seq_len_source,
                                               max_seq_len_target=max_seq_len_target,
                                               batch_size=batch_size,
                                               fill_up=fill_up)

    return train_iter, validation_iter, config_data, data_info


class LengthStatistics(config.Config):

    def __init__(self,
                 num_sents: int,
                 length_ratio_mean: float,
                 length_ratio_std: float) -> None:
        super().__init__()
        self.num_sents = num_sents
        self.length_ratio_mean = length_ratio_mean
        self.length_ratio_std = length_ratio_std


class DataStatistics(config.Config):

    def __init__(self,
                 num_sents: int,
                 num_discarded,

import mxnet as mx

from sockeye import __version__
from sockeye.config import Config
from . import constants as C
from . import data_io
from . import decoder
from . import encoder
from . import layers
from . import loss
from . import utils

logger = logging.getLogger(__name__)


class ModelConfig(Config):
    """
    ModelConfig defines model parameters defined at training time which are relevant to model inference.
    Add new model parameters here. If you want backwards compatibility for models trained with code that did not
    contain these parameters, provide a reasonable default under default_values.

    :param config_data: Used training data.
    :param vocab_source_size: Source vocabulary size.
    :param vocab_target_size: Target vocabulary size.
    :param config_embed_source: Embedding config for source.
    :param config_embed_target: Embedding config for target.
    :param config_encoder: Encoder configuration.
    :param config_decoder: Decoder configuration.
    :param config_loss: Loss configuration.
    :param weight_tying: Enables weight tying if True.
    :param weight_tying_type: Determines which weights get tied. Must be set if weight_tying is enabled.
    :param lhuc: LHUC (Vilar 2018) is applied at some part of the model.

# ...one iterator to traverse them all,
    scoring_iter = BatchedRawParallelSampleIter(data_loader=data_loader,
                                                sources=sources,
                                                target=target,
                                                source_vocabs=source_vocabs,
                                                target_vocab=target_vocab,
                                                bucket=bucket,
                                                batch_size=batch_size,
                                                max_lens=(max_seq_len_source, max_seq_len_target),
                                                num_factors=len(sources))

    # and with the model appraise them.
    return scoring_iter


class LengthStatistics(config.Config):

    def __init__(self,
                 num_sents: int,
                 length_ratio_mean: float,
                 length_ratio_std: float) -> None:
        super().__init__()
        self.num_sents = num_sents
        self.length_ratio_mean = length_ratio_mean
        self.length_ratio_std = length_ratio_std


class DataStatistics(config.Config):

    def __init__(self,
                 num_sents: int,
                 num_discarded,

"""
import logging
from typing import Callable

import mxnet as mx

from . import config
from . import constants as C
from . import layers
from . import rnn
from . import utils

logger = logging.getLogger(__name__)


class CoverageConfig(config.Config):
    """
    Coverage configuration.

    :param type: Coverage name.
    :param num_hidden: Number of hidden units for coverage networks.
    :param layer_normalization: Apply layer normalization to coverage networks.
    :param max_fertility: Maximum number of target words generated by a source word.
    """
    def __init__(self,
                 type: str,
                 num_hidden: int,
                 layer_normalization: bool,
                 max_fertility: int = 2) -> None:
        super().__init__()
        self.type = type
        self.max_fertility = max_fertility

import math
from abc import abstractmethod
from collections import namedtuple
from typing import Any, Dict, Optional, Tuple

import mxnet as mx

from . import config
from .lr_scheduler import LearningRateScheduler
from .utils import check_condition

BatchState = namedtuple("BatchState", ["metric_val"])
CheckpointState = namedtuple("CheckpointState", ["checkpoint", "metric_val"])


class OptimizerConfig(config.Config):

    def __init__(self,
                 name: str,
                 params: Dict[str, Any],
                 kvstore: str,
                 initializer: mx.initializer.Initializer,
                 gradient_clipping_type: str,
                 gradient_clipping_threshold: Optional[float],
                 update_interval: int = 1) -> None:
        super().__init__()
        self.name = name
        self.params = params
        self.kvstore = kvstore
        self.initializer = initializer
        self.gradient_clipping_type = gradient_clipping_type
        self.gradient_clipping_threshold = gradient_clipping_threshold

bias = utils.cast_conditionally(bias, self.dtype)
        for i, layer in enumerate(self.layers):
            # (batch_size, seq_len, config.model_size)
            data = layer(data, bias)
        data = self.final_process(data=data, prev=None)
        data = utils.uncast_conditionally(data, self.dtype)
        return data, data_length, seq_len

    def get_num_hidden(self) -> int:
        """
        Return the representation size of this encoder.
        """
        return self.config.model_size


class ConvolutionalEmbeddingConfig(config.Config):
    """
    Convolutional embedding encoder configuration.

    :param num_embed: Input embedding size.
    :param output_dim: Output segment embedding size.
    :param max_filter_width: Maximum filter width for convolutions.
    :param num_filters: Number of filters of each width.
    :param pool_stride: Stride for pooling layer after convolutions.
    :param num_highway_layers: Number of highway layers for segment embeddings.
    :param dropout: Dropout probability.
    :param add_positional_encoding: Dropout probability.
    :param dtype: Data type.
    """

    def __init__(self,
                 num_embed: int,

return get_transformer_encoder(config, prefix)
    elif isinstance(config, ConvolutionalEncoderConfig):
        return get_convolutional_encoder(config, prefix)
    elif isinstance(config, EmptyEncoderConfig):
        return EncoderSequence([EmptyEncoder(config)], config.dtype)
    else:
        from .image_captioning.encoder import ImageLoadedCnnEncoderConfig, \
            get_image_cnn_encoder

        if isinstance(config, ImageLoadedCnnEncoderConfig):
            return get_image_cnn_encoder(config)
        else:
            raise ValueError("Unsupported encoder configuration")


class RecurrentEncoderConfig(config.Config):
    """
    Recurrent encoder configuration.

    :param rnn_config: RNN configuration.
    :param conv_config: Optional configuration for convolutional embedding.
    :param reverse_input: Reverse embedding sequence before feeding into RNN.
    :param dtype: Data type.
    """

    def __init__(self,
                 rnn_config: rnn.RNNConfig,
                 conv_config: Optional['ConvolutionalEmbeddingConfig'] = None,
                 reverse_input: bool = False,
                 dtype: str = C.DTYPE_FP32) -> None:
        super().__init__()
        self.rnn_config = rnn_config