HA custom component for controlling Keenetic 3G internet link with ZTE MF833 modem

configuration.yaml

switch: 
  - platform: modem_control
    name: Megafon Internet
    host: !secret router_ip
    username: !secret router_username
    password: !secret router_password

custom_components|modem_control|__init__.py

"""Integrates Modem control."""

custom_components|modem_control|manifest.json

{
  "domain": "modem_control",
  "name": "Modem control",
  "documentation": "https://gist.github.com/foxel/01360210ba2b9a012bed3f586a8918cc",
  "requirements": ["ndms2_client"],
  "codeowners": ["@foxel"]
}

custom_components|modem_control|switch.py

"""Support for a switch to turn Modem on/off."""
from datetime import timedelta
import logging
from typing import Callable, List

from ndms2_client import TelnetConnection
from ndms2_client.client import _parse_dict_lines
import requests
import voluptuous as vol

from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
from homeassistant.const import (
    CONF_HOST,
    CONF_NAME,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_USERNAME,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
import homeassistant.util.dt as dt_util

_LOGGER = logging.getLogger(__name__)

CONF_INTERFACE = "interface"
DEFAULT_NAME = "Modem Switch"

DEFAULT_INTERFACE = "CdcEthernet0"
DEFAULT_PORT = 23

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Required(CONF_USERNAME): cv.string,
        vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
        vol.Required(CONF_PASSWORD): cv.string,
        vol.Required(CONF_NAME, default=DEFAULT_NAME): cv.string,
        vol.Required(CONF_INTERFACE, default=DEFAULT_INTERFACE): cv.string,
    }
)

DOMAIN = "modem_control"


def setup_platform(
    hass: HomeAssistantType,
    config: ConfigType,
    add_entities: Callable[[List[Entity]], None],
    _discovery_info=None,
):
    """Find and return switches controlled by shell commands."""

    setup_reload_service(hass, DOMAIN, ["switch"])

    add_entities([ModemSwitch(config)])


class ModemSwitch(SwitchEntity):
    """Representation a switch that can be toggled using shell commands."""

    def __init__(self, config: ConfigType):
        """Initialize the switch."""
        # self.entity_id = ENTITY_ID_FORMAT.format("modem_switch")
        self._name = config.get(CONF_NAME, DEFAULT_NAME)
        self._state = False
        self._state_attributes = {}
        self._conn = TelnetConnection(
            config[CONF_HOST],
            config.get(CONF_PORT, DEFAULT_PORT),
            config[CONF_USERNAME],
            config[CONF_PASSWORD],
        )
        self._interface = config.get(CONF_INTERFACE, DEFAULT_INTERFACE)
        self.update()

    @property
    def should_poll(self):
        """Only poll if we have state command."""
        return True

    @property
    def name(self):
        """Return the name of the switch."""
        return self._name

    @property
    def is_on(self):
        """Return true if device is on."""
        return self._state

    @property
    def state_attributes(self):
        """Return self._state_attributes."""
        return self._state_attributes

    @property
    def assumed_state(self):
        """Return true if we do optimistic updates."""
        return False

    def update(self):
        """Update device state."""
        result = _parse_dict_lines(
            self._router_command(f"show interface {self._interface}")
        )

        self._state_attributes = result
        self._state = result.get("state") == "up"

    def turn_on(self, **kwargs):
        """Turn the device on."""
        self._router_command(f"interface {self._interface} up")

        def set_manual_dial(_now):
            modem_ip = self._get_modem_ip()
            self._modem_command(
                modem_ip, "SET_CONNECTION_MODE", {"ConnectionMode": "manual_dial"}
            )

        async_track_point_in_utc_time(
            self.hass,
            set_manual_dial,
            dt_util.utcnow() + timedelta(seconds=20),
        )

        self.schedule_update_ha_state()

    def turn_off(self, **kwargs):
        """Turn the device off."""
        modem_ip = self._get_modem_ip()
        self._modem_command(
            modem_ip, "SET_CONNECTION_MODE", {"ConnectionMode": "manual_dial"}
        )
        self._modem_command(modem_ip, "DISCONNECT_NETWORK")
        self._router_command(f"interface {self._interface} down")

        self.schedule_update_ha_state()

    def _router_command(self, command) -> List[str]:
        result = self._conn.run_command(command)
        self._conn.disconnect()
        return result

    def _get_modem_ip(self) -> str:
        from ndms2_client import Client

        client = Client(self._conn)
        ips = [
            device.ip
            for device in client.get_arp_devices()
            if device.interface == self._interface
        ]
        self._conn.disconnect()
        return ips.pop(0)

    def _modem_command(self, ip: str, goform_id: str, params: dict = {}):
        response = requests.get(
            f"http://{ip}/goform/goform_set_cmd_process",
            params={**params, "goformId": goform_id},
            headers={"Referer": f"http://{ip}/index.html"},
        )
        assert response.json()["result"] == "success", f"Failed {goform_id}"