text: from typing import Mapping
  
  from libp2p.network.connection.raw_connection_interface import IRawConnection
  from libp2p.network.typing import GenericProtocolHandlerFn
  from libp2p.peer.id import ID
  from libp2p.protocol_muxer.exceptions import MultiselectClientError, MultiselectError
  from libp2p.security.exceptions import HandshakeFailure
  from libp2p.security.secure_conn_interface import ISecureConn
  from libp2p.security.secure_transport_interface import ISecureTransport
  from libp2p.security.security_multistream import SecurityMultistream
  from libp2p.stream_muxer.abc import IMuxedConn
  from libp2p.stream_muxer.muxer_multistream import MuxerClassType, MuxerMultistream
  from libp2p.transport.exceptions import MuxerUpgradeFailure, SecurityUpgradeFailure
  from libp2p.typing import TProtocol
  
  from .listener_interface import IListener
  from .transport_interface import ITransport
  
  
  class TransportUpgrader:
      security_multistream: SecurityMultistream
      muxer_multistream: MuxerMultistream
  
      def __init__(
          self,
          secure_transports_by_protocol: Mapping[TProtocol, ISecureTransport],
          muxer_transports_by_protocol: Mapping[TProtocol, MuxerClassType],
      ):
          self.security_multistream = SecurityMultistream(secure_transports_by_protocol)
          self.muxer_multistream = MuxerMultistream(muxer_transports_by_protocol)
  
      def upgrade_listener(self, transport: ITransport, listeners: IListener) -> None:
          """
          Upgrade multiaddr listeners to libp2p-transport listeners
          """
          # TODO: Figure out what to do with this function.
          pass
  
      async def upgrade_security(
          self, raw_conn: IRawConnection, peer_id: ID, initiator: bool
      ) -> ISecureConn:
          """
          Upgrade conn to a secured connection
          """
          try:
              if initiator:
                  return await self.security_multistream.secure_outbound(
                      raw_conn, peer_id
                  )
              return await self.security_multistream.secure_inbound(raw_conn)
          except (MultiselectError, MultiselectClientError) as error:
              raise SecurityUpgradeFailure(
                  "failed to negotiate the secure protocol"
              ) from error
          except HandshakeFailure as error:
              raise SecurityUpgradeFailure(
                  "handshake failed when upgrading to secure connection"
              ) from error
  
      async def upgrade_connection(
          self,
          conn: ISecureConn,
          generic_protocol_handler: GenericProtocolHandlerFn,
          peer_id: ID,
      ) -> IMuxedConn:
          """
          Upgrade secured connection to a muxed connection
          """
          try:
              return await self.muxer_multistream.new_conn(
                  conn, generic_protocol_handler, peer_id
              )
          except (MultiselectError, MultiselectClientError) as error:
              raise MuxerUpgradeFailure(
                  "failed to negotiate the multiplexer protocol"
              ) from error
  