text: # Copyright 2021 Johannes Verherstraeten
  #
  # 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.
  
  from typing import TypeVar, Generic, Sequence, TYPE_CHECKING, Optional
  from threading import BoundedSemaphore
  
  from pypipeline.cell.icell import ICell
  from pypipeline.cell.icellobserver import IObservable
  from pypipeline.cell.compositecell.acompositecell import ACompositeCell
  from pypipeline.cell.compositecell.scalablecell.strategy.cloningstrategy.clonecell.iclonecell import ICloneCell
  from pypipeline.exceptions import InvalidStateException, NotDeployedException, InvalidInputException, \
      CannotBeDeletedException
  from pypipeline.cellio import InternalInput, InternalOutput, IConnectionExitPoint, IConnectionEntryPoint
  from pypipeline.validation import BoolExplained, FalseExplained, TrueExplained, raise_if_not
  
  if TYPE_CHECKING:
      from pypipeline.cell.compositecell.icompositecell import ICompositeCell
      from pypipeline.cellio import IInput, IOutput
  
  
  T = TypeVar('T')
  
  
  class CloneInput(InternalInput[T], Generic[T]):
      """
      CloneInput class.
  
      A clone input is a type of internal input, and gets automatically generated by clone cells.
      It accepts no incoming connections and infinite outgoing (internal) connections.
  
      An IO is owned by its cell.
  
      An IO is the controlling class in the IO-ICell relation, as IO of the cell.
      An IConnectionExitPoint is the controlled class in the IConnection-IConnectionExitPoint relation, as the
      source of the connection.
      """
  
      def __init__(self, cell: "ICloneCell", name: str):
          super(CloneInput, self).__init__(cell, name)
  
  
  class CloneOutput(InternalOutput[T], Generic[T]):
      """
      CloneOutput class.
  
      A clone output is a type of internal output, and gets automatically generated by clone cells.
      It accepts 1 incoming (internal) connection and no outgoing connections.
  
      An IO is owned by its cell.
  
      An IO is the controlling class in the IO-ICell relation, as IO of the cell.
      An IConnectionEntryPoint is the controlled class in the IConnection-IConnectionEntryPoint relation, as the
      target of the connection.
      """
  
      PULL_TIMEOUT: float = 5.
  
      def __init__(self, cell: "ICloneCell", name: str):
          super(CloneOutput, self).__init__(cell, name)
          self.__value_is_set: BoundedSemaphore = BoundedSemaphore(1)
          self.__value_is_set.acquire()
  
      def set_value(self, value: T) -> None:
          super(CloneOutput, self).set_value(value)
          self.__value_is_set.release()
  
      def pull(self) -> T:
          if self.get_nb_incoming_connections() == 0:
              while not self.__value_is_set.acquire(timeout=self.PULL_TIMEOUT):
                  self.logger.info(f"{self} waiting for the clone to set its outputs...")
                  if not self._is_deployed():
                      raise NotDeployedException(f"{self} is pulled while not deployed")
              return self.get_value()
          return super(CloneOutput, self).pull()
  
  
  class ACloneCell(ACompositeCell, ICloneCell):
      """
      Abstract clone cell class.
  
      Controlling class in the IObserver-IObservable relation, as observer of the original cell.
      -> Relation should be created/deleted in the __init__ and delete() method.
      """
  
      def __init__(self, original_cell: "ICell", name: str, max_nb_internal_cells: int = 1):
          """
          Args:
              original_cell: the original cell that is cloned.
              name: the name of this clone cell.
              max_nb_internal_cells: max amount of internal cells this clone cell can have. Usually 0 or 1.
          Raises:
              InvalidInputException
          """
          super(ACloneCell, self).__init__(None, name, max_nb_internal_cells=max_nb_internal_cells,
                                           wants_to_observe_its_internal_cells=False)
          self.logger.info(f"creating clone cell")
  
          # Main mutator in the IObserver-IObservable relation, as observer of the original cell.
          raise_if_not(self.can_have_as_original_cell(original_cell), InvalidInputException)
          original_cell._add_observer(self)  # Access to protected method on purpose
          self.__original_cell = original_cell
  
          for original_input in original_cell.get_inputs():
              if isinstance(original_input, IConnectionEntryPoint):
                  input_name = original_input.get_name()
                  CloneInput(self, input_name)
  
          for original_output in original_cell.get_outputs():
              if isinstance(original_output, IConnectionExitPoint):
                  output_name = original_output.get_name()
                  CloneOutput(self, output_name)
  
      @classmethod
      def create(cls, original_cell: "ICell", name: str) -> "ACloneCell":
          raise NotImplementedError
  
      # ------ Original cell ------
  
      def has_as_observable(self, observable: "IObservable") -> bool:
          return super(ACloneCell, self).has_as_observable(observable) or self.get_original_cell() == observable
  
      def get_original_cell(self) -> "ICell":
          return self.__original_cell
  
      def can_have_as_original_cell(self, original_cell: "ICell") -> "BoolExplained":
          if not isinstance(original_cell, ICell):
              return FalseExplained(f"{self}: original cell should be of type ICell, got {type(original_cell)}. ")
          return TrueExplained()
  
      def assert_has_proper_original_cell(self) -> None:
          raise_if_not(self.can_have_as_original_cell(self.get_original_cell()), InvalidStateException)
          if not self.get_original_cell().has_as_observer(self):
              raise InvalidStateException(f"Inconsistent relation: {self.get_original_cell()} doesn't have {self} "
                                          f"as observer. ")
  
      # ------ Inputs & Outputs ------
  
      def get_clone_inputs(self) -> Sequence[CloneInput]:
          return [input_ for input_ in self.get_inputs() if isinstance(input_, CloneInput)]
  
      def get_clone_input(self, name: str) -> CloneInput:
          result: IInput = self.get_input(name)
          if not isinstance(result, CloneInput):
              raise KeyError
          return result
  
      def get_clone_outputs(self) -> Sequence[CloneOutput]:
          return [output for output in self.get_outputs() if isinstance(output, CloneOutput)]
  
      def get_clone_output(self, name: str) -> CloneOutput:
          result: IOutput = self.get_output(name)
          if not isinstance(result, CloneOutput):
              raise KeyError
          return result
  
      # ------ Other methods ------
  
      def _on_pull(self) -> None:
          raise NotImplementedError
  
      def clone(self, new_parent: "Optional[ICompositeCell]") -> "ICloneCell":
          raise Exception()   # should never be called
  
      def supports_scaling(self) -> bool:
          raise Exception()   # should never be called
  
      def get_nb_available_pulls(self) -> Optional[int]:
          raise Exception("A clone has no knowledge of the number of times it can be pulled.")
  
      def assert_is_valid(self) -> None:
          super(ACloneCell, self).assert_is_valid()
          self.assert_has_proper_original_cell()
  
      def delete(self) -> None:
          raise_if_not(self.can_be_deleted(), CannotBeDeletedException)
          # Main mutator in the IObserver-IObservable relation, as observer of the original cell.
          self.__original_cell._remove_observer(self)
          self.__original_cell = None
          super(ACloneCell, self).delete()
  
      def __str__(self) -> str:
          return f"{self.__class__.__name__}({self.get_name()})"
  