Enable strict typing for asterisk_cdr + asterisk_mbox (#106841)

This commit is contained in:
Marc Mueller 2024-01-01 20:45:16 +01:00 committed by GitHub
parent 3b0d877b5e
commit 8501b2e71b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 53 additions and 18 deletions

View File

@ -81,6 +81,8 @@ homeassistant.components.aranet.*
homeassistant.components.aruba.*
homeassistant.components.aseko_pool_live.*
homeassistant.components.assist_pipeline.*
homeassistant.components.asterisk_cdr.*
homeassistant.components.asterisk_mbox.*
homeassistant.components.asuswrt.*
homeassistant.components.auth.*
homeassistant.components.automation.*

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import datetime
import hashlib
from typing import Any
from homeassistant.components.asterisk_mbox import (
DOMAIN as ASTERISK_DOMAIN,
@ -28,21 +29,21 @@ async def async_get_handler(
class AsteriskCDR(Mailbox):
"""Asterisk VM Call Data Record mailbox."""
def __init__(self, hass, name):
def __init__(self, hass: HomeAssistant, name: str) -> None:
"""Initialize Asterisk CDR."""
super().__init__(hass, name)
self.cdr = []
self.cdr: list[dict[str, Any]] = []
async_dispatcher_connect(self.hass, SIGNAL_CDR_UPDATE, self._update_callback)
@callback
def _update_callback(self, msg):
def _update_callback(self, msg: list[dict[str, Any]]) -> Any:
"""Update the message count in HA, if needed."""
self._build_message()
self.async_update()
def _build_message(self):
def _build_message(self) -> None:
"""Build message structure."""
cdr = []
cdr: list[dict[str, Any]] = []
for entry in self.hass.data[ASTERISK_DOMAIN].cdr:
timestamp = datetime.datetime.strptime(
entry["time"], "%Y-%m-%d %H:%M:%S"
@ -61,7 +62,7 @@ class AsteriskCDR(Mailbox):
cdr.append({"info": info, "sha": sha, "text": msg})
self.cdr = cdr
async def async_get_messages(self):
async def async_get_messages(self) -> list[dict[str, Any]]:
"""Return a list of the current messages."""
if not self.cdr:
self._build_message()

View File

@ -1,5 +1,6 @@
"""Support for Asterisk Voicemail interface."""
import logging
from typing import Any, cast
from asterisk_mbox import Client as asteriskClient
from asterisk_mbox.commands import (
@ -42,11 +43,11 @@ CONFIG_SCHEMA = vol.Schema(
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up for the Asterisk Voicemail box."""
conf = config[DOMAIN]
conf: dict[str, Any] = config[DOMAIN]
host = conf[CONF_HOST]
port = conf[CONF_PORT]
password = conf[CONF_PASSWORD]
host: str = conf[CONF_HOST]
port: int = conf[CONF_PORT]
password: str = conf[CONF_PASSWORD]
hass.data[DOMAIN] = AsteriskData(hass, host, port, password, config)
@ -56,13 +57,20 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
class AsteriskData:
"""Store Asterisk mailbox data."""
def __init__(self, hass, host, port, password, config):
def __init__(
self,
hass: HomeAssistant,
host: str,
port: int,
password: str,
config: dict[str, Any],
) -> None:
"""Init the Asterisk data object."""
self.hass = hass
self.config = config
self.messages = None
self.cdr = None
self.messages: list[dict[str, Any]] | None = None
self.cdr: list[dict[str, Any]] | None = None
dispatcher_connect(self.hass, SIGNAL_MESSAGE_REQUEST, self._request_messages)
dispatcher_connect(self.hass, SIGNAL_CDR_REQUEST, self._request_cdr)
@ -71,7 +79,7 @@ class AsteriskData:
self.client = asteriskClient(host, port, password, self.handle_data)
@callback
def _discover_platform(self, component):
def _discover_platform(self, component: str) -> None:
_LOGGER.debug("Adding mailbox %s", component)
self.hass.async_create_task(
discovery.async_load_platform(
@ -80,10 +88,13 @@ class AsteriskData:
)
@callback
def handle_data(self, command, msg):
def handle_data(
self, command: int, msg: list[dict[str, Any]] | dict[str, Any]
) -> None:
"""Handle changes to the mailbox."""
if command == CMD_MESSAGE_LIST:
msg = cast(list[dict[str, Any]], msg)
_LOGGER.debug("AsteriskVM sent updated message list: Len %d", len(msg))
old_messages = self.messages
self.messages = sorted(
@ -93,6 +104,7 @@ class AsteriskData:
async_dispatcher_send(self.hass, SIGNAL_DISCOVER_PLATFORM, DOMAIN)
async_dispatcher_send(self.hass, SIGNAL_MESSAGE_UPDATE, self.messages)
elif command == CMD_MESSAGE_CDR:
msg = cast(dict[str, Any], msg)
_LOGGER.debug(
"AsteriskVM sent updated CDR list: Len %d", len(msg.get("entries", []))
)
@ -112,13 +124,13 @@ class AsteriskData:
)
@callback
def _request_messages(self):
def _request_messages(self) -> None:
"""Handle changes to the mailbox."""
_LOGGER.debug("Requesting message list")
self.client.messages()
@callback
def _request_cdr(self):
def _request_cdr(self) -> None:
"""Handle changes to the CDR."""
_LOGGER.debug("Requesting CDR list")
self.client.get_cdr()

View File

@ -74,7 +74,7 @@ class AsteriskMailbox(Mailbox):
async def async_get_messages(self) -> list[dict[str, Any]]:
"""Return a list of the current messages."""
data: AsteriskData = self.hass.data[ASTERISK_DOMAIN]
return data.messages
return data.messages or []
async def async_delete(self, msgid: str) -> bool:
"""Delete the specified messages."""

View File

@ -570,6 +570,26 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.asterisk_cdr.*]
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.asterisk_mbox.*]
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.asuswrt.*]
check_untyped_defs = true
disallow_incomplete_defs = true