jmq-olt-huawei 0.2.0

Paquete de Integración OLT Huawei MA56XXT con Python

README - jmq_olt_huawei

Paquete de integración con OLTs Huawei MA56XXT (como MA5603T) vía Telnet y SNMP asíncrono, diseñado para automatizar la recolección de información GPON desde Python.

🛰️ ¿Qué hace este paquete?

Permite conectarse a una OLT Huawei MA56XXT y ejecutar operaciones como:

• Listar slots activos (display board 0)
• Consultar puertos GPON por slot
• Obtener ONTs conectadas a cada puerto, incluyendo:
  • Potencia TX (ptx) y RX (prx) vía SNMP asíncrono
• Realizar un escaneo completo (slots → puertos → ONTs)
• Manejar paginación, prompts dinámicos y errores comunes de sesión (bloqueo de usuario)

---

📦 Instalación

Requiere Python ≥ 3.7 y PySNMP ≥ 6.2 (para usar el HLAPI asíncrono).

pip install jmq_olt_huawei

o bien:

pip install git+https://github.com/juaquicar/jmq_olt_huawei.git

Para desarrollo en modo editable:

pip install -e .

---

📁 Estructura del paquete

jmq_olt_huawei/
│
├── ma56xxt.py          # Lógica principal de conexión Telnet y SNMP asíncrono
├── __init__.py         # Archivo de inicialización del paquete
├── Examples/           # Scripts de ejemplo (opcional)
├── tests/              # Pruebas automatizadas (pytest)
├── requirements.txt    # Dependencias de desarrollo / CI
├── pyproject.toml      # Configuración de build
├── LICENSE             # Licencia MIT
└── README.md           # Este archivo

---

🧪 Ejemplo de uso

A continuación se muestra un ejemplo mínimo de cómo usar APIMA56XXT con llamadas asíncronas para obtener la lista completa de ONTs y sus potencias SNMP.

import asyncio
from jmq_olt_huawei.ma56xxt import APIMA56XXT, UserBusyError
from pprint import pprint

async def main():
    api = APIMA56XXT(
        host='192.168.88.25',
        user='root',
        password='admin',
        prompt='MA5603T',
        snmp_ip='192.168.88.25',
        snmp_port=161,
        snmp_community='public',
        debug=True
    )

    try:
        # Conexión Telnet (síncrona)
        api.connect()
        # Escaneo completo (async)
        full = await api.scan_all()
        pprint(full)
    except UserBusyError as e:
        print(f"ERROR: {e}")
    finally:
        api.close()  # Cierre de Telnet

if __name__ == '__main__':
    asyncio.run(main())

Flujo de ejemplo

1. api.connect() abre una sesión Telnet, hace login, entra en modo enable y config.

2. await api.scan_all() recorre:

   • get_slots() (síncrono, Telnet)

   • Para cada slot tipo GPBD: get_ports(slot) (síncrono)

   • Para cada puerto con optical_state == 'Online': await get_onts(slot, port_id) (asíncrono)

     • Cada ONT “online” dispara consultas SNMP asíncronas a await _snmp_potencia_tx(...) y await _snmp_potencia_rx(...).

3. Devuelve una lista jerárquica de slots → puertos → ONTs con campos SNMP (ptx, prx).

---

🔍 Manual de uso de los métodos de APIMA56XXT

1. __init__(host, user, password, prompt, snmp_ip, snmp_port, snmp_community, timeout=2.0, debug=False)

Inicializa la clase:

• host (str): IP o hostname de la OLT.
• user (str): Usuario Telnet.
• password (str): Contraseña Telnet.
• prompt (str): Prefijo del prompt (p.ej. "MA5603T").
• snmp_ip (str): IP de la OLT para SNMP.
• snmp_port (int): Puerto SNMP (por defecto 161).
• snmp_community (str): Comunidad SNMP (p.ej. "public").
• timeout (float): Segundos de espera en lecturas Telnet (por defecto 2.0).
• debug (bool): Si True, imprime logs de depuración.

api = APIMA56XXT(
    host='192.168.88.25',
    user='root',
    password='admin',
    prompt='MA5603T',
    snmp_ip='192.168.88.25',
    snmp_port=161,
    snmp_community='public',
    timeout=2.0,
    debug=True
)

---

2. connect()

Abre sesión Telnet, hace login y entra en modo operativo:

api.connect()
# Si debug=True:
# [DEBUG] Conectando a 192.168.88.25
# Conectado en modo config

• No devuelve valor.
• Lanza excepción (EOFError, socket.error) si la conexión falla.
• Después de llamar, se puede invocar get_slots(), get_ports(), etc.

---

3. disconnect()

Sale de los modos CLI (config, enable) y cierra la sesión Telnet:

api.disconnect()
# Si debug=True imprime:
# [DEBUG] Desconectado

• Protege contra llamadas dobles (si self.tn is None, no hace nada).

---

4. get_slots() → List[Tuple[str, str]]

Obtiene los slots en la OLT con display board 0:

slots = api.get_slots()
# Ejemplo:
# [('0', 'GPBD'), ('6', 'SCUN'), ('7', 'SCUN'), ('9', 'GICF')]

• Ejecuta internamente:

  1. _send('display board 0')
  2. _read_until_prompt()
  3. Parsea cada línea para extraer (slot_id, tipo).

• Es método síncrono (no necesita await).

---

5. get_ports(slot: str) → List[dict]

Obtiene la lista de puertos GPON de un slot específico:

ports = api.get_ports('0')
# Ejemplo de un diccionario en la lista:
# {
#   'id': 0,
#   'schema_fsp': '0/0/0',
#   'optical_state': 'Online',
#   'port_state': 'Offline',
#   'laser_state': 'Normal',
#   'bw': '1239040',
#   'temperature': '35',
#   'tx_bias': '12',
#   'voltage': '3.22',
#   'tx_power': '3.72',
#   'illegal_rogue_ont': 'Inexistent',
#   'max_distance': '40',
#   'wave_length': '1490',
#   'fiber_type': 'Single Mode',
#   'length': '-'
# }

• Paso a paso:

  1. _send(f'interface gpon 0/{slot}')
  2. _read_until_prompt()
  3. _send('display port state all')
  4. _read_until_prompt()
  5. Parsea bloques con _parse_port_block.
  6. _send('quit') + _read_until_prompt()

• Retorna lista de diccionarios, uno por puerto.

• Método síncrono (no usa await).

---

6. async get_onts(slot: str, port_id: int) → List[dict]

Devuelve la lista de ONTs conectadas a un puerto GPON:

onts = await api.get_onts('0', 0)
# Ejemplo de un diccionario ONT:
# {
#   'id': 2,
#   'schema_fsp': '0/0/0',
#   'sn': '48575443E010C4AC',
#   'control_flag': 'active',
#   'run_state': 'online',
#   'config_state': 'normal',
#   'match_state': 'match',
#   'protect_side': 'no',
#   'description': 'Cliente01',
#   'ptx': 2.34,      # Potencia TX (mW), float o "" si no está disponible
#   'prx': -21.57     # Potencia RX (dBm), float o "" si no está disponible
# }

• Internamente:

  1. _send(f'interface gpon 0/{slot}') (Telnet) + _read_until_prompt()

  2. _send(f'display ont info {port_id} all') + _read_until_prompt()

  3. _send('quit') + _read_until_prompt()

  4. Parsea sección principal y descripción con _parse_onts.

  5. Para cada ONT con run_state.lower() == 'online':

     • Llama a await _snmp_potencia_tx(slot, port_id, ont_id)
     • Llama a await _snmp_potencia_rx(slot, port_id, ont_id)

  6. Si SNMP falla, asigna "" a ptx y prx.

• Debe invocarse con await.

• Retorna lista de diccionarios ONT, incluyendo campos SNMP (ptx, prx).

---

7. async scan_all() → List[dict]

Realiza un escaneo completo de la OLT:

1. Obtiene todos los slots vía get_slots() (síncrono).

2. Para cada slot de tipo GPBD:

   • Llama a get_ports(slot) (síncrono).

   • Para cada puerto donde optical_state.lower() == 'online':

     • Llama a await get_onts(slot, port_id) (asíncrono).

   • Agrega un campo 'onts' a cada puerto (lista vacía si está offline).

3. Devuelve lista de diccionarios, cada uno así:

{
  "id": "0",
  "tipo": "GPBD",
  "ports": [
    {
      "id": 0,
      "schema_fsp": "0/0/0",
      "optical_state": "Online",
      "port_state": "Online",
      "laser_state": "Normal",
      "bw": "1239040",
      "temperature": "35",
      "tx_bias": "12",
      "voltage": "3.22",
      "tx_power": "3.72",
      "illegal_rogue_ont": "Inexistent",
      "max_distance": "40",
      "wave_length": "1490",
      "fiber_type": "Single Mode",
      "length": "-",
      "onts": [
        {
          "id": 2,
          "schema_fsp": "0/0/0",
          "sn": "48575443E010C4AC",
          "control_flag": "active",
          "run_state": "online",
          "config_state": "normal",
          "match_state": "match",
          "protect_side": "no",
          "description": "Cliente01",
          "ptx": 2.34,
          "prx": -21.57
        }
      ]
    },
    {
      "id": 1,
      "schema_fsp": "0/0/1",
      "optical_state": "Offline",
      "port_state": "Offline",
      // …otros campos
      "onts": []
    }
    // … más puertos
  ]
}

• Debe invocarse con await api.scan_all().
• Incluye SNMP asíncrono para ONTs “online”.

---

🛠️ Métodos y privados útiles

• _read_until_prompt() Lee líneas desde Telnet hasta detectar el prompt dinámico. Atiende paginación (More, ---- More) y errores de login acumulado (UserBusyError).

• _parse_port_block(slot, bloque) → dict | None Dado un bloque de texto de display port state all, extrae campos como id, optical_state, laser_state, bw, etc.

• _parse_onts(raw, slot, port_id) → List[dict] Dada la salida de display ont info {port_id} all, extrae la tabla de ONTs y sus descripciones. Devuelve lista de dicts con id, sn, run_state, etc.

• _calcular_codigo(slot, pon) → str Calcula la parte numérica base del OID SNMP:

  base = 4194304000
  código = base + pon_int * 256 + slot_int * 8192
  devuelve "<código>."
  

• “Helpers” SNMP asíncronos (async def)

  • _consultagetsnmp(codigo_oid: str) → int | None Crea un UdpTransportTarget con await UdpTransportTarget.create((ip, port), timeout=1, retries=0), luego hace await get_cmd(...).
  • _snmp_potencia_tx(slot, port, ont) → float | "" Llama a await _consultagetsnmp(...) con OID para potencia TX y procesa la respuesta (mW).
  • _snmp_potencia_rx(slot, port, ont) → float | "" Similar para potencia RX (dBm).

---

✅ Tests

Para ejecutar pruebas unitarias:

pytest -s tests/test_ma56xxt.py

• Asegúrate de tener un entorno de testing adecuado o mocks para Telnet/SNMP.
• Los tests deben validar parsing de puertos, ONTs y comportamiento SNMP asíncrono.

---

🧩 Funcionalidades destacadas

• Prompt dinámico configurable Detecta automáticamente prompts como MA5603T>, MA5603T(config)#, etc.
• Manejo de paginación Envía \n si detecta líneas con More, ---- More o Press 'Q'.
• Bloqueo de usuario Si la OLT informa “Reenter times have reached the upper limit”, lanza UserBusyError.
• SNMP asíncrono (PySNMP ≥ 6.2) Obtiene potencias TX/RX de cada ONT “online” de forma concurrente sin bloquear la lectura Telnet.
• Escaneo jerárquico (slots → puertos → ONTs) Permite obtener toda la información en una sola llamada a await scan_all().

---

📢 Contribuyendo

1. Haz un fork del repositorio.

2. Crea una rama:

   git checkout -b feature/nombre-de-funcionalidad
   

3. Realiza tus cambios, añade tests y actualiza este README si es necesario.

4. Haz commit y push a tu rama.

5. Abre un Pull Request describiendo los cambios propuestos.

---

⚖️ Licencia

MIT © Juanma Quijada

---

🔗 Enlaces

• Homepage / GitHub: https://github.com/juaquicar/jmq_olt_huawei
• PyPI: https://pypi.org/project/jmq-olt-huawei/