content: # -*- coding: utf-8 -*-
  # Copyright 2015 Spotify AB. All rights reserved.
  #
  # The contents of this file are 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.
  
  import ipaddress
  import json
  import os
  import re
  import tempfile
  import time
  import uuid
  
  # import stdlib
  from abc import abstractmethod
  from builtins import super
  from collections import defaultdict
  
  # import third party lib
  from typing import (
      Optional,
      Dict,
      List,
      Union,
      Any,
      cast,
      Callable,
      TypeVar,
  )
  
  from typing_extensions import (
      TypedDict,
      DefaultDict,
  )
  
  from netmiko import file_transfer
  from requests.exceptions import ConnectionError
  from netutils.config.compliance import diff_network_config
  from netutils.interface import canonical_interface_name
  
  import napalm.base.constants as c
  
  # import NAPALM Base
  import napalm.base.helpers
  from napalm.base import NetworkDriver
  from napalm.base.exceptions import CommandErrorException
  from napalm.base.exceptions import ConnectionException
  from napalm.base.exceptions import MergeConfigException
  from napalm.base.exceptions import ReplaceConfigException
  from napalm.base.helpers import as_number
  from napalm.base.helpers import generate_regex_or
  from napalm.base.netmiko_helpers import netmiko_args
  from napalm.base import models
  from napalm.nxapi_plumbing import Device as NXOSDevice
  from napalm.nxapi_plumbing import (
      NXAPIAuthError,
      NXAPIConnectionError,
      NXAPICommandError,
  )
  
  F = TypeVar("F", bound=Callable[..., Any])
  ShowIPInterfaceReturn = TypedDict(
      "ShowIPInterfaceReturn",
      {
          "intf-name": str,
          "prefix": str,
          "unnum-intf": str,
          "masklen": str,
      },
  )
  
  
  def ensure_netmiko_conn(func: F) -> F:
      """Decorator that ensures Netmiko connection exists."""
  
      def wrap_function(self, filename=None, config=None):  # type: ignore
          try:
              netmiko_object = self._netmiko_device
              if netmiko_object is None:
                  raise AttributeError()
          except AttributeError:
              device_type = c.NETMIKO_MAP[self.platform]
              netmiko_optional_args = self.netmiko_optional_args
              if "port" in netmiko_optional_args:
                  netmiko_optional_args["port"] = 22
              self._netmiko_open(
                  device_type=device_type, netmiko_optional_args=netmiko_optional_args
              )
          func(self, filename=filename, config=config)
  
      return cast(F, wrap_function)
  
  
  class NXOSDriverBase(NetworkDriver):
      """Common code shared between nx-api and nxos_ssh."""
  
      def __init__(
          self,
          hostname: str,
          username: str,
          password: str,
          timeout: int = 60,
          optional_args: Optional[Dict] = None,
      ) -> None:
          if optional_args is None:
              optional_args = {}
          self.hostname = hostname
          self.username = username
          self.password = password
          self.timeout = timeout
          self.replace = True
          self.loaded = False
          self.merge_candidate = ""
          self.candidate_cfg = "candidate_config.txt"
          self.rollback_cfg = "rollback_config.txt"
          self._dest_file_system = optional_args.pop("dest_file_system", "bootflash:")
          self.force_no_enable = optional_args.get("force_no_enable", False)
          self.netmiko_optional_args = netmiko_args(optional_args)
          self.device: Optional[NXOSDevice] = None
  
      @ensure_netmiko_conn
      def load_replace_candidate(
          self, filename: Optional[str] = None, config: Optional[str] = None
      ) -> None:
          if not filename and not config:
              raise ReplaceConfigException(
                  "filename or config parameter must be provided."
              )
  
          if not filename:
              assert isinstance(config, str)
              tmp_file = self._create_tmp_file(config)
              filename = tmp_file
          else:
              if not os.path.isfile(filename):
                  raise ReplaceConfigException("File {} not found".format(filename))
  
          try:
              transfer_result = file_transfer(
                  self._netmiko_device,
                  source_file=filename,
                  dest_file=self.candidate_cfg,
                  file_system=self._dest_file_system,
                  direction="put",
                  overwrite_file=True,
              )
              if not transfer_result["file_exists"]:
                  raise ValueError()
          except Exception as e:
              msg = (
                  "Could not transfer file. There was an error "
                  "during transfer. Please make sure remote "
                  f"permissions are set. Caught Error:\n{repr(str(e))}"
              )
              raise ReplaceConfigException(msg)
  
          self.replace = True
          self.loaded = True
          if config and os.path.isfile(tmp_file):
              os.remove(tmp_file)
  
      def load_merge_candidate(
          self, filename: Optional[str] = None, config: Optional[str] = None
      ) -> None:
          if not filename and not config:
              raise MergeConfigException("filename or config param must be provided.")
  
          self.merge_candidate += "\n"  # insert one extra line
          if filename is not None:
              with open(filename, "r") as f:
                  self.merge_candidate += f.read()
          else:
              assert isinstance(config, str)
              self.merge_candidate += config
          self.replace = False
          self.loaded = True
  
      def _send_command(
          self, command: str, raw_text: bool = False
      ) -> Dict[str, Union[str, Dict[str, Any]]]:
          raise NotImplementedError
  
      def _check_file_exists(self, cfg_file: str) -> bool:
          """
          Check that the file exists on remote device using full path.
  
          cfg_file can be a full path, e.g.: bootflash:rollback_config.txt
          or just a filename, e.g.: rollback_config.txt
  
          For example
          # dir rollback_config.txt
              71803    Sep 06 14:13:33 2023  rollback_config.txt
  
          Usage for bootflash://sup-local
          6211682304 bytes used
          110314684416 bytes free
          116526366720 bytes total
          """
          cmd = f"dir {cfg_file}"
          output = self._send_command(command=cmd, raw_text=True)
          if "No such file or directory" in output:
              return False
          else:
              return True
  
      def _commit_merge(self) -> None:
          try:
              output = self._send_config(self.merge_candidate)
              if output and "Invalid command" in output:
                  raise MergeConfigException("Error while applying config!")
          except Exception as e:
              self.rollback()
              err_header = "Configuration merge failed; automatic rollback attempted"
              merge_error = "{0}:\n{1}".format(err_header, repr(str(e)))
              raise MergeConfigException(merge_error)
  
          # clear the merge buffer
          self.merge_candidate = ""
  
      def _get_merge_diff(self) -> str:
          """
          Uses netutils diff_network_config to create a partial configuration
          with the proper hierarchy.
          Note: the netutils utility performs the diff offline.
  
          Returns: diff with the proper hierarchy of commands
          that are missing from the current config.
          Examples:
          Candidate configuration:
          interface loopback0
            ip address 10.1.4.5/32
            ip router ospf 100 area 0.0.0.1
  
          Base (on device) - relevant section:
          ...
          interface loopback0
            ip address 10.1.4.4/32
            ip router ospf 100 area 0.0.0.1
          ...
  
          Diff that respects the required command hierarchy:
          interface loopback0
            ip address 10.1.4.5/32
          """
          running_config = self.get_config(retrieve="running", full=True)["running"]
          return diff_network_config(self.merge_candidate, running_config, "cisco_nxos")
  
      def _get_diff(self) -> str:
          """Get a diff between running config and a proposed file."""
          diff: List[str] = []
          self._create_sot_file()
          diff_out = self._send_command(
              "show diff rollback-patch file {} file {}".format(
                  "sot_file", self.candidate_cfg
              ),
              raw_text=True,
          )
          assert isinstance(diff_out, str)
          try:
              diff_out = (
                  diff_out.split("Generating Rollback Patch")[1]
                  .replace("Rollback Patch is Empty", "")
                  .strip()
              )
              for line in diff_out.splitlines():
                  if line:
                      if line[0].strip() != "!" and line[0].strip() != ".":
                          diff.append(line.rstrip(" "))
          except (AttributeError, KeyError):
              raise ReplaceConfigException(
                  "Could not calculate diff. It's possible the given file doesn't exist."
              )
          return "\n".join(diff)
  
      def compare_config(self) -> str:
          if self.loaded:
              if not self.replace:
                  return self._get_merge_diff()
              diff = self._get_diff()
              return diff
          return ""
  
      def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None:
          if revert_in is not None:
              raise NotImplementedError(
                  "Commit confirm has not been implemented on this platform."
              )
          if message:
              raise NotImplementedError(
                  "Commit message not implemented for this platform"
              )
          if self.loaded:
              # Create checkpoint from current running-config
              self._save_to_checkpoint(self.rollback_cfg)
  
              if self.replace:
                  self._load_cfg_from_checkpoint()
              else:
                  self._commit_merge()
  
              try:
                  # If hostname changes ensure Netmiko state is updated properly
                  self._netmiko_device.set_base_prompt()
              except AttributeError:
                  pass
  
              self._copy_run_start()
              self.loaded = False
          else:
              raise ReplaceConfigException("No config loaded.")
  
      def discard_config(self) -> None:
          if self.loaded:
              # clear the buffer
              self.merge_candidate = ""
          if self.loaded and self.replace:
              self._delete_file(self.candidate_cfg)
          self.loaded = False
  
      def _create_sot_file(self) -> None:
          """Create Source of Truth file to compare."""
  
          # Bug on on NX-OS 6.2.16 where overwriting sot_file would take exceptionally long time
          # (over 12 minutes); so just delete the sot_file
          try:
              self._delete_file(filename="sot_file")
          except Exception:
              pass
          commands = [
              "terminal dont-ask",
              "checkpoint file sot_file",
              "no terminal dont-ask",
          ]
          self._send_command_list(commands)
  
      def ping(
          self,
          destination: str,
          source: str = c.PING_SOURCE,
          ttl: int = c.PING_TTL,
          timeout: int = c.PING_TIMEOUT,
          size: int = c.PING_SIZE,
          count: int = c.PING_COUNT,
          vrf: str = c.PING_VRF,
          source_interface: str = c.PING_SOURCE_INTERFACE,
      ) -> models.PingResultDict:
          """
          Execute ping on the device and returns a dictionary with the result.
          Output dictionary has one of following keys:
              * success
              * error
          In case of success, inner dictionary will have the followin keys:
              * probes_sent (int)
              * packet_loss (int)
              * rtt_min (float)
              * rtt_max (float)
              * rtt_avg (float)
              * rtt_stddev (float)
              * results (list)
          'results' is a list of dictionaries with the following keys:
              * ip_address (str)
              * rtt (float)
          """
          ping_dict: models.PingResultDict = {}
  
          version = ""
          try:
              version = "6" if ipaddress.ip_address(destination).version == 6 else ""
          except ValueError:
              # Allow use of DNS names
              pass
  
          command = "ping{version} {destination}".format(
              version=version, destination=destination
          )
          command += " timeout {}".format(timeout)
          command += " packet-size {}".format(size)
          command += " count {}".format(count)
          if source != "":
              command += " source {}".format(source)
          elif source_interface != "":
              command += " source {}".format(source_interface)
  
          if vrf != "":
              command += " vrf {}".format(vrf)
          output = self._send_command(command, raw_text=True)
          assert isinstance(output, str)
  
          if "connect:" in output:
              ping_dict["error"] = output
          elif "PING" in output:
              ping_dict["success"] = {
                  "probes_sent": 0,
                  "packet_loss": 0,
                  "rtt_min": 0.0,
                  "rtt_max": 0.0,
                  "rtt_avg": 0.0,
                  "rtt_stddev": 0.0,
                  "results": [],
              }
              results_array = []
              for line in output.splitlines():
                  fields = line.split()
                  if "icmp" in line:
                      if "Unreachable" in line:
                          if "(" in fields[2]:
                              results_array.append(
                                  {"ip_address": str(fields[2][1:-1]), "rtt": 0.0}
                              )
                          else:
                              results_array.append(
                                  {"ip_address": str(fields[1]), "rtt": 0.0}
                              )
                      elif "truncated" in line:
                          if "(" in fields[4]:
                              results_array.append(
                                  {"ip_address": str(fields[4][1:-2]), "rtt": 0.0}
                              )
                          else:
                              results_array.append(
                                  {"ip_address": str(fields[3][:-1]), "rtt": 0.0}
                              )
                      elif fields[1] == "bytes":
                          if version == "6":
                              m = fields[5][5:]
                          else:
                              m = fields[6][5:]
                          results_array.append(
                              {"ip_address": str(fields[3][:-1]), "rtt": float(m)}
                          )
                  elif "packets transmitted" in line:
                      ping_dict["success"]["probes_sent"] = int(fields[0])
                      ping_dict["success"]["packet_loss"] = int(fields[0]) - int(
                          fields[3]
                      )
                  elif "min/avg/max" in line:
                      split_fields = fields[3].split("/")
                      ping_dict["success"].update(
                          {
                              "rtt_min": float(split_fields[0]),
                              "rtt_avg": float(split_fields[1]),
                              "rtt_max": float(split_fields[2]),
                          }
                      )
              ping_dict["success"].update({"results": results_array})
          return ping_dict
  
      def traceroute(
          self,
          destination: str,
          source: str = c.TRACEROUTE_SOURCE,
          ttl: int = c.TRACEROUTE_TTL,
          timeout: int = c.TRACEROUTE_TIMEOUT,
          vrf: str = c.TRACEROUTE_VRF,
      ) -> models.TracerouteResultDict:
          _HOP_ENTRY_PROBE = [
              r"\s+",
              r"(",  # beginning of host_name (ip_address) RTT group
              r"(",  # beginning of host_name (ip_address) group only
              r"([a-zA-Z0-9\.:-]*)",  # hostname
              r"\s+",
              r"\(?([a-fA-F0-9\.:][^\)]*)\)?",  # IP Address between brackets
              r"(?:\s+\(AS\s+[0-9]+\))?",  # AS number -- may or may not be present
              r")?",  # end of host_name (ip_address) group only
              # also hostname/ip are optional -- they can or cannot be specified
              # if not specified, means the current probe followed the same path as the previous
              r"\s+",
              r"(\d+\.\d+)\s+ms",  # RTT
              r"|\*",  # OR *, when non responsive hop
              r")",  # end of host_name (ip_address) RTT group
          ]
  
          _HOP_ENTRY = [r"\s?", r"(\d+)"]  # space before hop index?  # hop index
  
          traceroute_result: models.TracerouteResultDict = {}
          timeout = 5  # seconds
          probes = 3  # 3 probes/jop and this cannot be changed on NXOS!
  
          version = ""
          try:
              version = "6" if ipaddress.ip_address(destination).version == 6 else ""
          except ValueError:
              # Allow use of DNS names
              pass
  
          if source:
              source_opt = "source {source}".format(source=source)
              command = "traceroute{version} {destination} {source_opt}".format(
                  version=version, destination=destination, source_opt=source_opt
              )
          else:
              command = "traceroute{version} {destination}".format(
                  version=version, destination=destination
              )
  
          if vrf != "":
              command += " vrf {vrf}".format(vrf=vrf)
  
          try:
              traceroute_raw_output = self._send_command(command, raw_text=True)
          except CommandErrorException:
              return {
                  "error": "Cannot execute traceroute on the device: {}".format(command)
              }
  
          assert isinstance(traceroute_raw_output, str)
  
          hop_regex = "".join(_HOP_ENTRY + _HOP_ENTRY_PROBE * probes)
          traceroute_result["success"] = {}
          if traceroute_raw_output:
              for line in traceroute_raw_output.splitlines():
                  hop_search = re.search(hop_regex, line)
                  if not hop_search:
                      continue
                  hop_details = hop_search.groups()
                  hop_index = int(hop_details[0])
                  previous_probe_host_name = "*"
                  previous_probe_ip_address = "*"
                  traceroute_result["success"][hop_index] = {"probes": {}}
                  for probe_index in range(probes):
                      host_name = hop_details[3 + probe_index * 5]
                      ip_address_raw = hop_details[4 + probe_index * 5]
                      ip_address = napalm.base.helpers.convert(
                          napalm.base.helpers.ip, ip_address_raw, ip_address_raw
                      )
                      rtt_as_string = hop_details[5 + probe_index * 5]
                      if rtt_as_string:
                          rtt = float(rtt_as_string)
                      else:
                          rtt = timeout * 1000.0
                      if not host_name:
                          host_name = previous_probe_host_name
                      if not ip_address:
                          ip_address = previous_probe_ip_address
                      if hop_details[1 + probe_index * 5] == "*":
                          host_name = "*"
                          ip_address = "*"
                      traceroute_result["success"][hop_index]["probes"][
                          probe_index + 1
                      ] = {
                          "host_name": str(host_name),
                          "ip_address": str(ip_address),
                          "rtt": rtt,
                      }
                      previous_probe_host_name = host_name
                      previous_probe_ip_address = ip_address
          return traceroute_result
  
      def _get_checkpoint_file(self) -> str:
          filename = "temp_cp_file_from_napalm"
          self._set_checkpoint(filename)
          command = "show file {}".format(filename)
          output = self._send_command(command, raw_text=True)
          assert isinstance(output, str)
          self._delete_file(filename)
          return output
  
      def _set_checkpoint(self, filename: str) -> None:
          commands = [
              "terminal dont-ask",
              "checkpoint file {}".format(filename),
              "no terminal dont-ask",
          ]
          self._send_command_list(commands)
  
      def _save_to_checkpoint(self, filename: str) -> None:
          """Save the current running config to the given file."""
          commands = [
              "terminal dont-ask",
              "checkpoint file {}".format(filename),
              "no terminal dont-ask",
          ]
          self._send_command_list(commands)
  
      def _delete_file(self, filename: str) -> None:
          commands = [
              "terminal dont-ask",
              "delete {}".format(filename),
              "no terminal dont-ask",
          ]
          self._send_command_list(commands)
  
      @staticmethod
      def _create_tmp_file(config: str) -> str:
          tmp_dir = tempfile.gettempdir()
          rand_fname = str(uuid.uuid4())
          filename = os.path.join(tmp_dir, rand_fname)
          with open(filename, "wt") as fobj:
              fobj.write(config)
          return filename
  
      def _disable_confirmation(self) -> None:
          self._send_command_list(["terminal dont-ask"])
  
      def get_config(
          self,
          retrieve: str = "all",
          full: bool = False,
          sanitized: bool = False,
          format: str = "text",
      ) -> models.ConfigDict:
          # NX-OS adds some extra, unneeded lines that should be filtered.
          filter_strings = [
              r"!Command: show .*$",
              r"!Time:.*\d{4}\s*$",
              r"Startup config saved at:.*$",
          ]
          filter_pattern = generate_regex_or(filter_strings)
  
          config: models.ConfigDict = {
              "startup": "",
              "running": "",
              "candidate": "",
          }  # default values
          # NX-OS only supports "all" on "show run"
          run_full = " all" if full else ""
  
          if retrieve.lower() in ("running", "all"):
              command = f"show running-config{run_full}"
              output = self._send_command(command, raw_text=True)
              assert isinstance(output, str)
              output = re.sub(filter_pattern, "", output, flags=re.M)
              config["running"] = output.strip()
          if retrieve.lower() in ("startup", "all"):
              command = "show startup-config"
              output = self._send_command(command, raw_text=True)
              assert isinstance(output, str)
              output = re.sub(filter_pattern, "", output, flags=re.M)
              config["startup"] = output.strip()
  
          if sanitized:
              return napalm.base.helpers.sanitize_configs(
                  config, c.CISCO_SANITIZE_FILTERS
              )
  
          return config
  
      def get_lldp_neighbors(self) -> Dict[str, List[models.LLDPNeighborDict]]:
          """IOS implementation of get_lldp_neighbors."""
          lldp: Dict[str, List[models.LLDPNeighborDict]] = {}
          neighbors_detail = self.get_lldp_neighbors_detail()
          for intf_name, entries in neighbors_detail.items():
              lldp[intf_name] = []
              for lldp_entry in entries:
                  hostname = lldp_entry["remote_system_name"]
                  # Match IOS behaviour of taking remote chassis ID
                  # When lacking a system name (in show lldp neighbors)
                  if hostname == "N/A":
                      hostname = lldp_entry["remote_chassis_id"]
                  lldp_dict: models.LLDPNeighborDict = {
                      "port": lldp_entry["remote_port"],
                      "hostname": hostname,
                  }
                  lldp[intf_name].append(lldp_dict)
  
          return lldp
  
      def get_lldp_neighbors_detail(
          self, interface: str = ""
      ) -> models.LLDPNeighborsDetailDict:
          lldp: models.LLDPNeighborsDetailDict = {}
          lldp_interfaces: List[str] = []
  
          if interface:
              command = "show lldp neighbors interface {} detail".format(interface)
          else:
              command = "show lldp neighbors detail"
          lldp_entries = self._send_command(command, raw_text=True)
          lldp_entries = str(lldp_entries)
          lldp_entries = napalm.base.helpers.textfsm_extractor(
              self, "show_lldp_neighbors_detail", lldp_entries
          )
  
          if len(lldp_entries) == 0:
              return {}
  
          for idx, lldp_entry in enumerate(lldp_entries):
              local_intf = lldp_entry.pop("local_interface") or lldp_interfaces[idx]
              # Convert any 'not advertised' to an empty string
              for field in lldp_entry:
                  if "not advertised" in lldp_entry[field]:
                      lldp_entry[field] = ""
              # Add field missing on IOS
              lldp_entry["parent_interface"] = ""
              # Translate the capability fields
              lldp_entry["remote_system_capab"] = (
                  napalm.base.helpers.transform_lldp_capab(
                      lldp_entry["remote_system_capab"]
                  )
              )
              lldp_entry["remote_system_enable_capab"] = (
                  napalm.base.helpers.transform_lldp_capab(
                      lldp_entry["remote_system_enable_capab"]
                  )
              )
              # Turn the interfaces into their long version
              local_intf = canonical_interface_name(local_intf)
              lldp.setdefault(local_intf, [])
              lldp[local_intf].append(lldp_entry)  # type: ignore
  
          return lldp
  
      @staticmethod
      def _get_table_rows(
          parent_table: Optional[Dict], table_name: str, row_name: str
      ) -> List:
          """
          Inconsistent behavior:
          {'TABLE_intf': [{'ROW_intf': {
          vs
          {'TABLE_mac_address': {'ROW_mac_address': [{
          vs
          {'TABLE_vrf': {'ROW_vrf': {'TABLE_adj': {'ROW_adj': {
          """
          if parent_table is None:
              return []
          _table = parent_table.get(table_name)
          _table_rows = []
          if isinstance(_table, list):
              _table_rows = [_table_row.get(row_name) for _table_row in _table]
          elif isinstance(_table, dict):
              _table_rows = _table.get(row_name)  # type: ignore
          if not isinstance(_table_rows, list):
              _table_rows = [_table_rows]
          return _table_rows
  
      def _get_reply_table(
          self, result: Optional[Dict], table_name: str, row_name: str
      ) -> List:
          return self._get_table_rows(result, table_name, row_name)
  
      def _get_command_table(self, command: str, table_name: str, row_name: str) -> List:
          json_output = self._send_command(command)
          if type(json_output) is not dict and json_output:
              assert isinstance(json_output, str)
              json_output = json.loads(json_output)
          return self._get_reply_table(json_output, table_name, row_name)
  
      def _parse_vlan_ports(self, vlan_s: Union[str, List]) -> List:
          vlans = []
          find_regexp = r"^([A-Za-z\/-]+|.*\/)(\d+)-(\d+)$"
          vlan_str = ""
  
          if isinstance(vlan_s, list):
              vlan_str = ",".join(vlan_s)
          else:
              vlan_str = vlan_s
  
          for vls in vlan_str.split(","):
              find = re.findall(find_regexp, vls.strip())
              if find:
                  for i in range(int(find[0][1]), int(find[0][2]) + 1):
                      vlans.append(canonical_interface_name(find[0][0] + str(i)))
              else:
                  vlans.append(canonical_interface_name(vls.strip()))
          return vlans
  
      @abstractmethod
      def _send_config(self, commands: Union[str, List]) -> List[str]:
          raise NotImplementedError
  
      @abstractmethod
      def _load_cfg_from_checkpoint(self) -> None:
          raise NotImplementedError
  
      @abstractmethod
      def _copy_run_start(self) -> None:
          raise NotImplementedError
  
      @abstractmethod
      def _send_command_list(self, commands: List[str]) -> List[str]:
          raise NotImplementedError
  
  
  class NXOSDriver(NXOSDriverBase):
      def __init__(
          self,
          hostname: str,
          username: str,
          password: str,
          timeout: int = 60,
          optional_args: Optional[Dict] = None,
      ) -> None:
          super().__init__(
              hostname, username, password, timeout=timeout, optional_args=optional_args
          )
          if optional_args is None:
              optional_args = {}
  
          # nxos_protocol is there for backwards compatibility, transport is the preferred method
          self.transport = optional_args.get(
              "transport", optional_args.get("nxos_protocol", "https")
          )
          if self.transport == "https":
              self.port = optional_args.get("port", 443)
          elif self.transport == "http":
              self.port = optional_args.get("port", 80)
  
          self.ssl_verify = optional_args.get("ssl_verify", False)
          self.platform = "nxos"
  
      def open(self) -> None:
          try:
              self.device = NXOSDevice(
                  host=self.hostname,
                  username=self.username,
                  password=self.password,
                  timeout=self.timeout,
                  port=self.port,
                  transport=self.transport,
                  verify=self.ssl_verify,
                  api_format="jsonrpc",
              )
              self._send_command("show hostname")
          except (NXAPIConnectionError, NXAPIAuthError):
              # unable to open connection
              raise ConnectionException("Cannot connect to {}".format(self.hostname))
  
      def close(self) -> None:
          self.device = None
  
      def _send_command(self, command: str, raw_text: bool = False) -> Any:
          """
          Wrapper for NX-API show method.
  
          Allows more code sharing between NX-API and SSH.
          """
          assert (
              self.device is not None
          ), "Call open() or use as a context manager before calling _send_command"
          return self.device.show(command, raw_text=raw_text)
  
      def _send_command_list(self, commands: List[str]) -> List[Any]:
          assert (
              self.device is not None
          ), "Call open() or use as a context manager before running _send_command_list"
          return self.device.config_list(commands)
  
      def _send_config(self, commands: Union[str, List]) -> List[str]:
          assert (
              self.device is not None
          ), "Call open() or use as a context manager before running _send_config"
          if isinstance(commands, str):
              # Has to be a list generator and not generator expression (not JSON serializable)
              commands = [command for command in commands.splitlines() if command]
          return self.device.config_list(commands)
  
      @staticmethod
      def _compute_timestamp(stupid_cisco_output: str) -> float:
          """
          Some fields such `uptime` are returned as: 23week(s) 3day(s)
          This method will determine the epoch of the event.
          e.g.: 23week(s) 3day(s) -> 1462248287
          """
          if not stupid_cisco_output or stupid_cisco_output == "never":
              return -1.0
  
          if "(s)" in stupid_cisco_output:
              pass
          elif ":" in stupid_cisco_output:
              stupid_cisco_output = stupid_cisco_output.replace(":", "hour(s) ", 1)
              stupid_cisco_output = stupid_cisco_output.replace(":", "minute(s) ", 1)
              stupid_cisco_output += "second(s)"
          else:
              stupid_cisco_output = stupid_cisco_output.replace("d", "day(s) ")
              stupid_cisco_output = stupid_cisco_output.replace("h", "hour(s)")
  
          things: Dict[str, Dict[str, Union[int, float]]] = {
              "second(s)": {"weight": 1},
              "minute(s)": {"weight": 60},
              "hour(s)": {"weight": 3600},
              "day(s)": {"weight": 24 * 3600},
              "week(s)": {"weight": 7 * 24 * 3600},
              "year(s)": {"weight": 365.25 * 24 * 3600},
          }
  
          things_keys = things.keys()
          for part in stupid_cisco_output.split():
              for key in things_keys:
                  if key in part:
                      things[key]["count"] = napalm.base.helpers.convert(
                          int, part.replace(key, "")
                      )
  
          delta = sum([det.get("count", 0) * det["weight"] for det in things.values()])
          return time.time() - delta
  
      def is_alive(self) -> models.AliveDict:
          if self.device:
              return {"is_alive": True}
          else:
  