Your First Real Device Automation Project

import csv
import json
import getpass
from netmiko import ConnectHandler, NetmikoTimeoutException, NetmikoAuthenticationException


def is_valid_device(device):
  """A device record needs at least a hostname and an IP to be usable."""
  return bool(device.get("hostname")) and bool(device.get("ip"))


def is_ntp_synced(show_ntp_output):
  """Check whether 'show ntp status' output indicates the clock is synchronized."""
  first_line = show_ntp_output.strip().splitlines()[0] if show_ntp_output.strip() else ""
  return first_line.startswith("Clock is synchronized")


def is_vlan_present(show_vlan_output, vlan_id):
  """Check whether a given VLAN ID appears in 'show vlan brief' output."""
  for line in show_vlan_output.splitlines():
      stripped = line.strip()
      first_word = stripped.split(" ")[0] if stripped else ""
      if first_word == str(vlan_id):
          return True
  return False


def connect_and_check(device, username, password):
  """
  Connects to a real device over SSH (read-only) and returns its NTP
  and VLAN compliance state.
  """
  try:
      connection = ConnectHandler(
          device_type=device["platform"],
          host=device["ip"],
          username=username,
          password=password,
          conn_timeout=10,
      )
  except NetmikoTimeoutException:
      raise ConnectionError(f"{device['hostname']} did not respond (timeout)")
  except NetmikoAuthenticationException:
      raise ConnectionError(f"{device['hostname']} rejected the credentials (authentication failure)")

  ntp_output = connection.send_command("show ntp status")
  vlan_output = connection.send_command("show vlan brief")
  connection.disconnect()

  return {
      "reachable": True,
      "ntp_synced": is_ntp_synced(ntp_output),
      "vlan_compliant": is_vlan_present(vlan_output, 20),
  }


def check_compliance(state):
  """Turn a device's raw state into a plain-language compliance status."""
  if state["ntp_synced"]:
      if state["vlan_compliant"]:
          return "compliant"
      else:
          return "vlan non-compliant"
  else:
      return "ntp non-compliant"


with open("devices.csv") as f:
  devices = list(csv.DictReader(f))

valid_devices = []
for device in devices:
  if is_valid_device(device):
      valid_devices.append(device)
  else:
      print(f"Skipping invalid inventory record: {device}")

username = input("Username: ")
password = getpass.getpass("Password: ")

results = []

for device in valid_devices:
  print(f"Checking {device['hostname']}...")

  try:
      state = connect_and_check(device, username, password)
  except ConnectionError as e:
      print(f"{device['hostname']}: {e}")
      results.append({
          "hostname": device["hostname"],
          "ip": device["ip"],
          "status": "unreachable",
      })
      continue

  status = check_compliance(state)
  print(f"{device['hostname']}: {status}")
  results.append({
      "hostname": device["hostname"],
      "ip": device["ip"],
      "status": status,
  })

with open("compliance_report.json", "w") as f:
  json.dump(results, f, indent=4)

compliant = 0
needs_attention = 0
for result in results:
  if result["status"] == "compliant":
      compliant += 1
  else:
      needs_attention += 1

print()
print("--- Summary ---")
print(f"Checked:         {len(results)} devices")
print(f"Compliant:       {compliant}")
print(f"Needs attention: {needs_attention}")
print("Report written to compliance_report.json")

Every function here was built across this module and the last one — nothing here is new to this lesson. devices.csv keeps the exact hostname/ip/platform format from Phase 2.6; credentials are never in that file.