mirror of
https://github.com/home-assistant/core.git
synced 2025-07-29 18:28:14 +02:00
@ -170,6 +170,7 @@ homeassistant.components.mailbox.*
|
|||||||
homeassistant.components.media_player.*
|
homeassistant.components.media_player.*
|
||||||
homeassistant.components.media_source.*
|
homeassistant.components.media_source.*
|
||||||
homeassistant.components.metoffice.*
|
homeassistant.components.metoffice.*
|
||||||
|
homeassistant.components.mikrotik.*
|
||||||
homeassistant.components.mjpeg.*
|
homeassistant.components.mjpeg.*
|
||||||
homeassistant.components.modbus.*
|
homeassistant.components.modbus.*
|
||||||
homeassistant.components.modem_callerid.*
|
homeassistant.components.modem_callerid.*
|
||||||
|
@ -1,15 +1,18 @@
|
|||||||
"""The Mikrotik component."""
|
"""The Mikrotik component."""
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
from homeassistant.const import Platform
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.exceptions import ConfigEntryNotReady
|
from homeassistant.exceptions import ConfigEntryNotReady
|
||||||
from homeassistant.helpers import config_validation as cv, device_registry as dr
|
from homeassistant.helpers import config_validation as cv, device_registry as dr
|
||||||
|
|
||||||
from .const import ATTR_MANUFACTURER, DOMAIN, PLATFORMS
|
from .const import ATTR_MANUFACTURER, DOMAIN
|
||||||
from .errors import CannotConnect, LoginError
|
from .errors import CannotConnect, LoginError
|
||||||
from .hub import MikrotikDataUpdateCoordinator, get_api
|
from .hub import MikrotikDataUpdateCoordinator, get_api
|
||||||
|
|
||||||
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
|
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
|
||||||
|
|
||||||
|
PLATFORMS = [Platform.DEVICE_TRACKER]
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
||||||
"""Set up the Mikrotik component."""
|
"""Set up the Mikrotik component."""
|
||||||
@ -26,7 +29,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
|
|||||||
|
|
||||||
hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator
|
hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator
|
||||||
|
|
||||||
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
|
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
|
||||||
|
|
||||||
device_registry = dr.async_get(hass)
|
device_registry = dr.async_get(hass)
|
||||||
device_registry.async_get_or_create(
|
device_registry.async_get_or_create(
|
||||||
|
@ -1,8 +1,6 @@
|
|||||||
"""Constants used in the Mikrotik components."""
|
"""Constants used in the Mikrotik components."""
|
||||||
from typing import Final
|
from typing import Final
|
||||||
|
|
||||||
from homeassistant.const import Platform
|
|
||||||
|
|
||||||
DOMAIN: Final = "mikrotik"
|
DOMAIN: Final = "mikrotik"
|
||||||
DEFAULT_NAME: Final = "Mikrotik"
|
DEFAULT_NAME: Final = "Mikrotik"
|
||||||
DEFAULT_API_PORT: Final = 8728
|
DEFAULT_API_PORT: Final = 8728
|
||||||
@ -40,7 +38,6 @@ MIKROTIK_SERVICES: Final = {
|
|||||||
IS_CAPSMAN: "/caps-man/interface/print",
|
IS_CAPSMAN: "/caps-man/interface/print",
|
||||||
}
|
}
|
||||||
|
|
||||||
PLATFORMS: Final = [Platform.DEVICE_TRACKER]
|
|
||||||
|
|
||||||
ATTR_DEVICE_TRACKER: Final = [
|
ATTR_DEVICE_TRACKER: Final = [
|
||||||
"comment",
|
"comment",
|
||||||
|
66
homeassistant/components/mikrotik/device.py
Normal file
66
homeassistant/components/mikrotik/device.py
Normal file
@ -0,0 +1,66 @@
|
|||||||
|
"""Network client device class."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from homeassistant.util import slugify
|
||||||
|
import homeassistant.util.dt as dt_util
|
||||||
|
|
||||||
|
from .const import ATTR_DEVICE_TRACKER
|
||||||
|
|
||||||
|
|
||||||
|
class Device:
|
||||||
|
"""Represents a network device."""
|
||||||
|
|
||||||
|
def __init__(self, mac: str, params: dict[str, Any]) -> None:
|
||||||
|
"""Initialize the network device."""
|
||||||
|
self._mac = mac
|
||||||
|
self._params = params
|
||||||
|
self._last_seen: datetime | None = None
|
||||||
|
self._attrs: dict[str, Any] = {}
|
||||||
|
self._wireless_params: dict[str, Any] = {}
|
||||||
|
|
||||||
|
@property
|
||||||
|
def name(self) -> str:
|
||||||
|
"""Return device name."""
|
||||||
|
return self._params.get("host-name", self.mac)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def ip_address(self) -> str | None:
|
||||||
|
"""Return device primary ip address."""
|
||||||
|
return self._params.get("address")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def mac(self) -> str:
|
||||||
|
"""Return device mac."""
|
||||||
|
return self._mac
|
||||||
|
|
||||||
|
@property
|
||||||
|
def last_seen(self) -> datetime | None:
|
||||||
|
"""Return device last seen."""
|
||||||
|
return self._last_seen
|
||||||
|
|
||||||
|
@property
|
||||||
|
def attrs(self) -> dict[str, Any]:
|
||||||
|
"""Return device attributes."""
|
||||||
|
attr_data = self._wireless_params | self._params
|
||||||
|
for attr in ATTR_DEVICE_TRACKER:
|
||||||
|
if attr in attr_data:
|
||||||
|
self._attrs[slugify(attr)] = attr_data[attr]
|
||||||
|
return self._attrs
|
||||||
|
|
||||||
|
def update(
|
||||||
|
self,
|
||||||
|
wireless_params: dict[str, Any] | None = None,
|
||||||
|
params: dict[str, Any] | None = None,
|
||||||
|
active: bool = False,
|
||||||
|
) -> None:
|
||||||
|
"""Update Device params."""
|
||||||
|
if wireless_params:
|
||||||
|
self._wireless_params = wireless_params
|
||||||
|
if params:
|
||||||
|
self._params = params
|
||||||
|
if active:
|
||||||
|
self._last_seen = dt_util.utcnow()
|
@ -63,7 +63,7 @@ def update_items(
|
|||||||
coordinator: MikrotikDataUpdateCoordinator,
|
coordinator: MikrotikDataUpdateCoordinator,
|
||||||
async_add_entities: AddEntitiesCallback,
|
async_add_entities: AddEntitiesCallback,
|
||||||
tracked: dict[str, MikrotikDataUpdateCoordinatorTracker],
|
tracked: dict[str, MikrotikDataUpdateCoordinatorTracker],
|
||||||
):
|
) -> None:
|
||||||
"""Update tracked device state from the hub."""
|
"""Update tracked device state from the hub."""
|
||||||
new_tracked: list[MikrotikDataUpdateCoordinatorTracker] = []
|
new_tracked: list[MikrotikDataUpdateCoordinatorTracker] = []
|
||||||
for mac, device in coordinator.api.devices.items():
|
for mac, device in coordinator.api.devices.items():
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
"""The Mikrotik router class."""
|
"""The Mikrotik router class."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import datetime, timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
import socket
|
import socket
|
||||||
import ssl
|
import ssl
|
||||||
@ -14,12 +14,9 @@ from homeassistant.config_entries import ConfigEntry
|
|||||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
|
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||||
from homeassistant.util import slugify
|
|
||||||
import homeassistant.util.dt as dt_util
|
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
ARP,
|
ARP,
|
||||||
ATTR_DEVICE_TRACKER,
|
|
||||||
ATTR_FIRMWARE,
|
ATTR_FIRMWARE,
|
||||||
ATTR_MODEL,
|
ATTR_MODEL,
|
||||||
ATTR_SERIAL_NUMBER,
|
ATTR_SERIAL_NUMBER,
|
||||||
@ -38,66 +35,12 @@ from .const import (
|
|||||||
NAME,
|
NAME,
|
||||||
WIRELESS,
|
WIRELESS,
|
||||||
)
|
)
|
||||||
|
from .device import Device
|
||||||
from .errors import CannotConnect, LoginError
|
from .errors import CannotConnect, LoginError
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class Device:
|
|
||||||
"""Represents a network device."""
|
|
||||||
|
|
||||||
def __init__(self, mac: str, params: dict[str, Any]) -> None:
|
|
||||||
"""Initialize the network device."""
|
|
||||||
self._mac = mac
|
|
||||||
self._params = params
|
|
||||||
self._last_seen: datetime | None = None
|
|
||||||
self._attrs: dict[str, Any] = {}
|
|
||||||
self._wireless_params: dict[str, Any] = {}
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self) -> str:
|
|
||||||
"""Return device name."""
|
|
||||||
return self._params.get("host-name", self.mac)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def ip_address(self) -> str | None:
|
|
||||||
"""Return device primary ip address."""
|
|
||||||
return self._params.get("address")
|
|
||||||
|
|
||||||
@property
|
|
||||||
def mac(self) -> str:
|
|
||||||
"""Return device mac."""
|
|
||||||
return self._mac
|
|
||||||
|
|
||||||
@property
|
|
||||||
def last_seen(self) -> datetime | None:
|
|
||||||
"""Return device last seen."""
|
|
||||||
return self._last_seen
|
|
||||||
|
|
||||||
@property
|
|
||||||
def attrs(self) -> dict[str, Any]:
|
|
||||||
"""Return device attributes."""
|
|
||||||
attr_data = self._wireless_params | self._params
|
|
||||||
for attr in ATTR_DEVICE_TRACKER:
|
|
||||||
if attr in attr_data:
|
|
||||||
self._attrs[slugify(attr)] = attr_data[attr]
|
|
||||||
return self._attrs
|
|
||||||
|
|
||||||
def update(
|
|
||||||
self,
|
|
||||||
wireless_params: dict[str, Any] | None = None,
|
|
||||||
params: dict[str, Any] | None = None,
|
|
||||||
active: bool = False,
|
|
||||||
) -> None:
|
|
||||||
"""Update Device params."""
|
|
||||||
if wireless_params:
|
|
||||||
self._wireless_params = wireless_params
|
|
||||||
if params:
|
|
||||||
self._params = params
|
|
||||||
if active:
|
|
||||||
self._last_seen = dt_util.utcnow()
|
|
||||||
|
|
||||||
|
|
||||||
class MikrotikData:
|
class MikrotikData:
|
||||||
"""Handle all communication with the Mikrotik API."""
|
"""Handle all communication with the Mikrotik API."""
|
||||||
|
|
||||||
@ -248,8 +191,8 @@ class MikrotikData:
|
|||||||
self, cmd: str, params: dict[str, Any] | None = None
|
self, cmd: str, params: dict[str, Any] | None = None
|
||||||
) -> list[dict[str, Any]]:
|
) -> list[dict[str, Any]]:
|
||||||
"""Retrieve data from Mikrotik API."""
|
"""Retrieve data from Mikrotik API."""
|
||||||
|
_LOGGER.debug("Running command %s", cmd)
|
||||||
try:
|
try:
|
||||||
_LOGGER.debug("Running command %s", cmd)
|
|
||||||
if params:
|
if params:
|
||||||
return list(self.api(cmd=cmd, **params))
|
return list(self.api(cmd=cmd, **params))
|
||||||
return list(self.api(cmd=cmd))
|
return list(self.api(cmd=cmd))
|
||||||
@ -273,7 +216,7 @@ class MikrotikData:
|
|||||||
return []
|
return []
|
||||||
|
|
||||||
|
|
||||||
class MikrotikDataUpdateCoordinator(DataUpdateCoordinator):
|
class MikrotikDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||||
"""Mikrotik Hub Object."""
|
"""Mikrotik Hub Object."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@ -293,7 +236,7 @@ class MikrotikDataUpdateCoordinator(DataUpdateCoordinator):
|
|||||||
@property
|
@property
|
||||||
def host(self) -> str:
|
def host(self) -> str:
|
||||||
"""Return the host of this hub."""
|
"""Return the host of this hub."""
|
||||||
return self.config_entry.data[CONF_HOST]
|
return str(self.config_entry.data[CONF_HOST])
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def hostname(self) -> str:
|
def hostname(self) -> str:
|
||||||
|
10
mypy.ini
10
mypy.ini
@ -1459,6 +1459,16 @@ disallow_untyped_defs = true
|
|||||||
warn_return_any = true
|
warn_return_any = true
|
||||||
warn_unreachable = true
|
warn_unreachable = true
|
||||||
|
|
||||||
|
[mypy-homeassistant.components.mikrotik.*]
|
||||||
|
check_untyped_defs = true
|
||||||
|
disallow_incomplete_defs = true
|
||||||
|
disallow_subclassing_any = true
|
||||||
|
disallow_untyped_calls = true
|
||||||
|
disallow_untyped_decorators = true
|
||||||
|
disallow_untyped_defs = true
|
||||||
|
warn_return_any = true
|
||||||
|
warn_unreachable = true
|
||||||
|
|
||||||
[mypy-homeassistant.components.mjpeg.*]
|
[mypy-homeassistant.components.mjpeg.*]
|
||||||
check_untyped_defs = true
|
check_untyped_defs = true
|
||||||
disallow_incomplete_defs = true
|
disallow_incomplete_defs = true
|
||||||
|
Reference in New Issue
Block a user