code: from typing import Optional
  
  from overrides import overrides
  import torch
  
  from allennlp.training.metrics.metric import Metric
  
  
  @Metric.register("entropy")
  class Entropy(Metric):
      def __init__(self) -> None:
          self._entropy = 0.0
          self._count = 0
  
      @overrides
      def __call__(
          self,  # type: ignore
          logits: torch.Tensor,
          mask: Optional[torch.Tensor] = None,
      ):
          """
          Parameters
          ----------
          logits : ``torch.Tensor``, required.
              A tensor of unnormalized log probabilities of shape (batch_size, ..., num_classes).
          mask: ``torch.Tensor``, optional (default = None).
              A masking tensor of shape (batch_size, ...).
          """
          logits, mask = self.unwrap_to_tensors(logits, mask)
  
          if mask is None:
              mask = torch.ones(logits.size()[:-1])
  
          log_probs = torch.nn.functional.log_softmax(logits, dim=-1)
          probabilities = torch.exp(log_probs) * mask.unsqueeze(-1)
          weighted_negative_likelihood = -log_probs * probabilities
          entropy = weighted_negative_likelihood.sum(-1)
  
          self._entropy += entropy.sum() / mask.sum()
          self._count += 1
  
      @overrides
      def get_metric(self, reset: bool = False):
          """
          Returns
          -------
          The scalar average entropy.
          """
          average_value = self._entropy / self._count if self._count > 0 else 0
          if reset:
              self.reset()
          return average_value
  
      @overrides
      def reset(self):
          self._entropy = 0.0
          self._count = 0
   import math
  
  from overrides import overrides
  import torch
  from torch.nn.parameter import Parameter
  
  from allennlp.modules.similarity_functions.similarity_function import SimilarityFunction
  from allennlp.nn import Activation, util
  
  
  @SimilarityFunction.register("linear")
  class LinearSimilarity(SimilarityFunction):
      """
      This similarity function performs a dot product between a vector of weights and some
      combination of the two input vectors, followed by an (optional) activation function.  The
      combination used is configurable.
  
      If the two vectors are ``x`` and ``y``, we allow the following kinds of combinations: ``x``,
      ``y``, ``x*y``, ``x+y``, ``x-y``, ``x/y``, where each of those binary operations is performed
      elementwise.  You can list as many combinations as you want, comma separated.  For example, you
      might give ``x,y,x*y`` as the ``combination`` parameter to this class.  The computed similarity
      function would then be ``w^T [x; y; x*y] + b``, where ``w`` is a vector of weights, ``b`` is a
      bias parameter, and ``[;]`` is vector concatenation.
  
      Note that if you want a bilinear similarity function with a diagonal weight matrix W, where the
      similarity function is computed as `x * w * y + b` (with `w` the diagonal of `W`), you can
      accomplish that with this class by using "x*y" for `combination`.
  
      Parameters
      ----------
      tensor_1_dim : ``int``
          The dimension of the first tensor, ``x``, described above.  This is ``x.size()[-1]`` - the
          length of the vector that will go into the similarity computation.  We need this so we can
          build weight vectors correctly.
      tensor_2_dim : ``int``
          The dimension of the second tensor, ``y``, described above.  This is ``y.size()[-1]`` - the
          length of the vector that will go into the similarity computation.  We need this so we can
          build weight vectors correctly.
      combination : ``str``, optional (default="x,y")
          Described above.
      activation : ``Activation``, optional (default=linear (i.e. no activation))
          An activation function applied after the ``w^T * [x;y] + b`` calculation.  Default is no
          activation.
      """
  
      def __init__(
          self,
          tensor_1_dim: int,
          tensor_2_dim: int,
          combination: str = "x,y",
          activation: Activation = None,
      ) -> None:
          super().__init__()
          self._combination = combination
          combined_dim = util.get_combined_dim(combination, [tensor_1_dim, tensor_2_dim])
          self._weight_vector = Parameter(torch.Tensor(combined_dim))
          self._bias = Parameter(torch.Tensor(1))
          self._activation = activation or Activation.by_name("linear")()
          self.reset_parameters()
  
      def reset_parameters(self):
          std = math.sqrt(6 / (self._weight_vector.size(0) + 1))
          self._weight_vector.data.uniform_(-std, std)
          self._bias.data.fill_(0)
  
      @overrides
      def forward(self, tensor_1: torch.Tensor, tensor_2: torch.Tensor) -> torch.Tensor:
          combined_tensors = util.combine_tensors(self._combination, [tensor_1, tensor_2])
          dot_product = torch.matmul(combined_tensors, self._weight_vector)
          return self._activation(dot_product + self._bias)
   from collections import Counter
  from multiprocessing import Queue, Process
  from queue import Empty
  from typing import Tuple
  
  import numpy as np
  
  from allennlp.common.testing import AllenNlpTestCase
  from allennlp.data.dataset_readers import MultiprocessDatasetReader, SequenceTaggingDatasetReader
  from allennlp.data.dataset_readers.multiprocess_dataset_reader import QIterable
  from allennlp.data.instance import Instance
  from allennlp.data.iterators import BasicIterator
  from allennlp.data.vocabulary import Vocabulary
  
  
  def fingerprint(instance: Instance) -> Tuple[str, ...]:
      """
      Get a hashable representation of a sequence tagging instance
      that can be put in a Counter.
      """
      text_tuple = tuple(t.text for t in instance.fields["tokens"].tokens)  # type: ignore
      labels_tuple = tuple(instance.fields["tags"].labels)  # type: ignore
      return text_tuple + labels_tuple
  
  
  class TestMultiprocessDatasetReader(AllenNlpTestCase):
      def setUp(self) -> None:
          super().setUp()
  
          # use SequenceTaggingDatasetReader as the base reader
          self.base_reader = SequenceTaggingDatasetReader(lazy=True)
          base_file_path = AllenNlpTestCase.FIXTURES_ROOT / "data" / "sequence_tagging.tsv"
  
          # Make 100 copies of the data
          raw_data = open(base_file_path).read()
          for i in range(100):
              file_path = self.TEST_DIR / f"identical_{i}.tsv"
              with open(file_path, "w") as f:
                  f.write(raw_data)
  
          self.all_distinct_path = str(self.TEST_DIR / "all_distinct.tsv")
          with open(self.all_distinct_path, "w") as all_distinct:
              for i in range(100):
                  file_path = self.TEST_DIR / f"distinct_{i}.tsv"
                  line = f"This###DT\tis###VBZ\tsentence###NN\t{i}###CD\t.###.\n"
                  with open(file_path, "w") as f:
                      f.write(line)
                  all_distinct.write(line)
  
          self.identical_files_glob = str(self.TEST_DIR / "identical_*.tsv")
          self.distinct_files_glob = str(self.TEST_DIR / "distinct_*.tsv")
  
          # For some of the tests we need a vocab, we'll just use the base_reader for that.
          self.vocab = Vocabulary.from_instances(self.base_reader.read(str(base_file_path)))
  
      def test_multiprocess_read(self):
          reader = MultiprocessDatasetReader(base_reader=self.base_reader, num_workers=4)
  
          all_instances = []
  
          for instance in reader.read(self.identical_files_glob):
              all_instances.append(instance)
  
          # 100 files * 4 sentences / file
          assert len(all_instances) == 100 * 4
  
          counts = Counter(fingerprint(instance) for instance in all_instances)
  
          # should have the exact same data 100 times
          assert len(counts) == 4
          assert counts[("cats", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("dogs", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("snakes", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("birds", "are", "animals", ".", "N", "V", "N", "N")] == 100
  
      def test_multiprocess_read_partial_does_not_hang(self):
          # Use a small queue size such that the processes generating the data will block.
          reader = MultiprocessDatasetReader(
              base_reader=self.base_reader, num_workers=4, output_queue_size=10
          )
  
          all_instances = []
  
          # Half of 100 files * 4 sentences / file
          i = 0
          for instance in reader.read(self.identical_files_glob):
              # Stop early such that the processes generating the data remain
              # active (given the small queue size).
              if i == 200:
                  break
              i += 1
              all_instances.append(instance)
  
          # This should be trivially true. The real test here is that we exit
          # normally and don't hang due to the still active processes.
          assert len(all_instances) == 200
  
      def test_multiprocess_read_with_qiterable(self):
          reader = MultiprocessDatasetReader(base_reader=self.base_reader, num_workers=4)
  
          all_instances = []
          qiterable = reader.read(self.identical_files_glob)
          assert isinstance(qiterable, QIterable)
  
          # Essentially QIterable.__iter__. Broken out here as we intend it to be
          # a public interface.
          qiterable.start()
          while qiterable.num_active_workers.value > 0 or qiterable.num_inflight_items.value > 0:
              while True:
                  try:
                      all_instances.append(qiterable.output_queue.get(block=False, timeout=1.0))
                      with qiterable.num_inflight_items.get_lock():
                          qiterable.num_inflight_items.value -= 1
                  except Empty:
                      break
          qiterable.join()
  
          # 100 files * 4 sentences / file
          assert len(all_instances) == 100 * 4
  
          counts = Counter(fingerprint(instance) for instance in all_instances)
  
          # should have the exact same data 100 times
          assert len(counts) == 4
          assert counts[("cats", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("dogs", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("snakes", "are", "animals", ".", "N", "V", "N", "N")] == 100
          assert counts[("birds", "are", "animals", ".", "N", "V", "N", "N")] == 100
  
      def test_multiprocess_read_in_subprocess_is_deterministic(self):
          reader = MultiprocessDatasetReader(base_reader=self.base_reader, num_workers=1)
          q = Queue()
  
          def read():
              for instance in reader.read(self.distinct_files_glob):
                  q.put(fingerprint(instance))
  
          # Ensure deterministic shuffling.
          np.random.seed(0)
          p = Process(target=read)
          p.start()
          p.join()
  
          # Convert queue to list.
          actual_fingerprints = []
          while not q.empty():
              actual_fingerprints.append(q.get(block=False))
  
          assert len(actual_fingerprints) == 100
  
          expected_fingerprints = []
          for instance in self.base_reader.read(self.all_distinct_path):
              expected_fingerprints.append(fingerprint(instance))
  
          np.random.seed(0)
          expected_fingerprints.sort()
          # This should be shuffled into exactly the same order as actual_fingerprints.
          np.random.shuffle(expected_fingerprints)
  
          assert actual_fingerprints == expected_fingerprints
  
      def test_multiple_epochs(self):
          reader = MultiprocessDatasetReader(
              base_reader=self.base_reader, num_workers=2, epochs_per_read=3
          )
  
          all_instances = []
  
          for instance in reader.read(self.identical_files_glob):
              all_instances.append(instance)
  
          # 100 files * 4 sentences per file * 3 epochs
          assert len(all_instances) == 100 * 4 * 3
  
          counts = Counter(fingerprint(instance) for instance in all_instances)
  
          # should have the exact same data 100 * 3 times
          assert len(counts) == 4
          assert counts[("cats", "are", "animals", ".", "N", "V", "N", "N")] == 300
          assert counts[("dogs", "are", "animals", ".", "N", "V", "N", "N")] == 300
          assert counts[("snakes", "are", "animals", ".", "N", "V", "N", "N")] == 300
          assert counts[("birds", "are", "animals", ".", "N", "V", "N", "N")] == 300
  
      def test_with_iterator(self):
          reader = MultiprocessDatasetReader(base_reader=self.base_reader, num_workers=2)
          instances = reader.read(self.identical_files_glob)
  
          iterator = BasicIterator(batch_size=32)
          iterator.index_with(self.vocab)
  
          batches = [batch for batch in iterator(instances, num_epochs=1)]
  
          # 400 instances / batch_size 32 = 12 full batches + 1 batch of 16
          sizes = sorted([len(batch["tags"]) for batch in batches])
          assert sizes == [16] + 12 * [32]
  
apis: torch.nn.functional.log_softmax torch.exp
apis: torch.Tensor torch.matmul
apis: numpy.random.shuffle numpy.random.seed