# -*- coding: utf-8 -*-
"""UPnP Server."""

# pylint: disable=too-many-lines

import asyncio
import logging
import socket
import sys
import time
import xml.etree.ElementTree as ET
from asyncio.transports import DatagramTransport
from datetime import datetime, timedelta, timezone
from functools import partial, wraps
from itertools import cycle
from random import randrange
from time import mktime
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Mapping,
    Optional,
    Sequence,
    Tuple,
    Type,
    Union,
    cast,
)
from urllib.parse import urlparse
from uuid import uuid4
from wsgiref.handlers import format_date_time

import defusedxml.ElementTree as DET  # pylint: disable=import-error
import voluptuous as vol
from aiohttp.web import (
    Application,
    AppRunner,
    HTTPBadRequest,
    Request,
    Response,
    RouteDef,
    TCPSite,
)

from async_upnp_client import __version__ as version
from async_upnp_client.aiohttp import AiohttpRequester
from async_upnp_client.client import (
    T,
    UpnpAction,
    UpnpDevice,
    UpnpError,
    UpnpRequester,
    UpnpService,
    UpnpStateVariable,
)
from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.const import (
    STATE_VARIABLE_TYPE_MAPPING,
    ActionArgumentInfo,
    ActionInfo,
    AddressTupleVXType,
    DeviceInfo,
    EventableStateVariableTypeInfo,
    HttpRequest,
    NotificationSubType,
    ServiceInfo,
    StateVariableInfo,
    StateVariableTypeInfo,
)
from async_upnp_client.exceptions import (
    UpnpActionError,
    UpnpActionErrorCode,
    UpnpValueError,
)
from async_upnp_client.ssdp import (
    _LOGGER_TRAFFIC_SSDP,
    SSDP_DISCOVER,
    SSDP_ST_ALL,
    SSDP_ST_ROOTDEVICE,
    SsdpProtocol,
    build_ssdp_packet,
    determine_source_target,
    get_ssdp_socket,
    is_ipv6_address,
)
from async_upnp_client.utils import CaseInsensitiveDict

NAMESPACES = {
    "s": "http://schemas.xmlsoap.org/soap/envelope/",
    "es": "http://schemas.xmlsoap.org/soap/encoding/",
}
HEADER_SERVER = f"async-upnp-client/{version} UPnP/2.0 Server/1.0"
HEADER_CACHE_CONTROL = "max-age=1800"
SSDP_SEARCH_RESPONDER_OPTIONS = "ssdp_search_responder_options"
SSDP_SEARCH_RESPONDER_OPTION_ALWAYS_REPLY_WITH_ROOT_DEVICE = (
    "ssdp_search_responder_always_rootdevice"
)
SSDP_SEARCH_RESPONDER_OPTION_HEADERS = "search_headers"
SSDP_ADVERTISEMENT_ANNOUNCER_OPTIONS = "ssdp_advertisement_announcer_options"
SSDP_ADVERTISEMENT_ANNOUNCER_OPTION_HEADERS = "advertisement_headers"

_LOGGER = logging.getLogger(__name__)
_LOGGER_TRAFFIC_UPNP = logging.getLogger("async_upnp_client.traffic.upnp")

# Hack: Bend INFO to DEBUG.
_LOGGER_TRAFFIC_UPNP.info = _LOGGER_TRAFFIC_UPNP.debug  # type: ignore


class NopRequester(UpnpRequester):  # pylint: disable=too-few-public-methods
    """NopRequester, does nothing."""


class EventSubscriber:
    """Represent a service subscriber."""

    DEFAULT_TIMEOUT = 3600

    def __init__(self, callback_url: str, timeout: Optional[int]) -> None:
        """Initialize."""
        self._url = callback_url
        self._uuid = str(uuid4())
        self._event_key = 0
        self._expires = datetime.now()
        self.timeout = timeout

    @property
    def url(self) -> str:
        """Return callback URL."""
        return self._url

    @property
    def uuid(self) -> str:
        """Return subscriber uuid."""
        return self._uuid

    @property
    def timeout(self) -> Optional[int]:
        """Return timeout in seconds."""
        return self._timeout

    @timeout.setter
    def timeout(self, timeout: Optional[int]) -> None:
        """Set timeout before unsubscribe."""
        if timeout is None:
            timeout = self.DEFAULT_TIMEOUT
        self._timeout = timeout
        self._expires = datetime.now() + timedelta(seconds=timeout)

    @property
    def expiration(self) -> datetime:
        """Return expiration time of subscription."""
        return self._expires

    def get_next_seq(self) -> int:
        """Return the next sequence number for an event."""
        res = self._event_key
        self._event_key += 1
        if self._event_key > 0xFFFF_FFFF:
            self._event_key = 1
        return res


class UpnpEventableStateVariable(UpnpStateVariable):
    """Representation of an eventable State Variable."""

    def __init__(
        self, state_variable_info: StateVariableInfo, schema: vol.Schema
    ) -> None:
        """Initialize."""
        super().__init__(state_variable_info, schema)
        self._last_sent = datetime.fromtimestamp(0, timezone.utc)
        self._defered_event: Optional[asyncio.TimerHandle] = None
        self._sent_event = asyncio.Event()

    @property
    def event_triggered(self) -> asyncio.Event:
        """Return event object for trigger completion."""
        return self._sent_event

    @property
    def max_rate(self) -> float:
        """Return max event rate."""
        type_info = cast(
            EventableStateVariableTypeInfo, self._state_variable_info.type_info
        )
        return type_info.max_rate or 0.0

    @property
    def value(self) -> Optional[T]:
        """Get Python value for this argument."""
        return super().value

    @value.setter
    def value(self, value: Any) -> None:
        """Set value, python typed."""
        if self._value == value:
            return
        super(UpnpEventableStateVariable, self.__class__).value.__set__(self, value)  # type: ignore
        if not self.service or self._defered_event:
            return
        assert self._updated_at
        next_update = self._last_sent + timedelta(seconds=self.max_rate)
        if self._updated_at >= next_update:
            asyncio.create_task(self.trigger_event())
        else:
            loop = asyncio.get_running_loop()
            self._defered_event = loop.call_at(
                next_update.timestamp(), self.trigger_event
            )

    async def trigger_event(self) -> None:
        """Update any waiting subscribers."""
        self._last_sent = datetime.now(timezone.utc)
        service = self.service
        assert isinstance(service, UpnpServerService)
        self._sent_event.set()
        asyncio.create_task(service.async_send_events())  # pylint: disable=no-member


class UpnpServerAction(UpnpAction):
    """Representation of an Action."""

    async def async_handle(self, **kwargs: Any) -> Any:
        """Handle action."""
        self.validate_arguments(**kwargs)
        raise NotImplementedError()


class UpnpServerService(UpnpService):
    """UPnP Service representation."""

    SERVICE_DEFINITION: ServiceInfo
    STATE_VARIABLE_DEFINITIONS: Mapping[str, StateVariableTypeInfo]

    def __init__(self, requester: UpnpRequester) -> None:
        """Initialize."""
        super().__init__(requester, self.SERVICE_DEFINITION, [], [])

        self._init_state_variables()
        self._init_actions()
        self._subscribers: List[EventSubscriber] = []

    def _init_state_variables(self) -> None:
        """Initialize state variables from STATE_VARIABLE_DEFINITIONS."""
        for name, type_info in self.STATE_VARIABLE_DEFINITIONS.items():
            self.create_state_var(name, type_info)

    def create_state_var(
        self, name: str, type_info: StateVariableTypeInfo
    ) -> UpnpStateVariable:
        """Create UpnpStateVariable."""
        existing = self.state_variables.get(name, None)
        if existing is not None:
            raise UpnpError(f"StateVariable with the same name exists: {name}")

        state_var_info = StateVariableInfo(
            name,
            send_events=isinstance(type_info, EventableStateVariableTypeInfo),
            type_info=type_info,
            xml=ET.Element("stateVariable"),
        )

        # pylint: disable=protected-access
        state_var: UpnpStateVariable
        if isinstance(type_info, EventableStateVariableTypeInfo):
            state_var = UpnpEventableStateVariable(
                state_var_info,
                UpnpFactory(self.requester)._state_variable_create_schema(type_info),
            )
        else:
            state_var = UpnpStateVariable(
                state_var_info,
                UpnpFactory(self.requester)._state_variable_create_schema(type_info),
            )
        state_var.service = self
        if type_info.default_value is not None:
            state_var.upnp_value = type_info.default_value

        self.state_variables[state_var.name] = state_var
        return state_var

    def _init_actions(self) -> None:
        """Initialize actions from annotated methods."""
        for item in dir(self):
            if item in ("control_url", "event_sub_url", "scpd_url", "device"):
                continue

            thing = getattr(self, item, None)
            if not thing or not hasattr(thing, "__upnp_action__"):
                continue

            self._init_action(thing)

    def _init_action(self, func: Callable) -> UpnpAction:
        """Initialize action for method."""
        name, in_args, out_args = cast(
            Tuple[str, Mapping[str, str], Mapping[str, str]],
            getattr(func, "__upnp_action__"),
        )

        arg_infos: List[ActionArgumentInfo] = []
        args: List[UpnpAction.Argument] = []
        for arg_name, state_var_name in in_args.items():
            # Validate function has parameter.
            assert arg_name in func.__annotations__

            # Validate parameter type.
            annotation = func.__annotations__.get(arg_name, None)
            state_var = self.state_variable(state_var_name)
            assert state_var.data_type_mapping["type"] == annotation

            # Build in-argument.
            arg_info = ActionArgumentInfo(
                arg_name,
                direction="in",
                state_variable_name=state_var.name,
                xml=ET.Element("server_argument"),
            )
            arg_infos.append(arg_info)

            arg = UpnpAction.Argument(arg_info, state_var)
            args.append(arg)

        for arg_name, state_var_name in out_args.items():
            # Build out-argument.
            state_var = self.state_variable(state_var_name)
            arg_info = ActionArgumentInfo(
                arg_name,
                direction="out",
                state_variable_name=state_var.name,
                xml=ET.Element("server_argument"),
            )
            arg_infos.append(arg_info)

            arg = UpnpAction.Argument(arg_info, state_var)
            args.append(arg)

        action_info = ActionInfo(
            name=name,
            arguments=arg_infos,
            xml=ET.Element("server_action"),
        )
        action = UpnpServerAction(action_info, args)
        action.async_handle = func  # type: ignore
        action.service = self
        self.actions[name] = action
        return action

    async def async_handle_action(self, action_name: str, **kwargs: Any) -> Any:
        """Handle action."""
        action = cast(UpnpServerAction, self.actions[action_name])
        action.validate_arguments(**kwargs)
        return await action.async_handle(**kwargs)

    def add_subscriber(self, subscriber: EventSubscriber) -> None:
        """Add or update a subscriber."""
        self._subscribers.append(subscriber)

    def del_subscriber(self, sid: str) -> bool:
        """Delete a subscriber."""
        subscriber = self.get_subscriber(sid)
        if subscriber:
            self._subscribers.remove(subscriber)
            return True
        return False

    def get_subscriber(self, sid: str) -> Optional[EventSubscriber]:
        """Get matching subscriber (if any)."""
        for subscriber in self._subscribers:
            if subscriber.uuid == sid:
                return subscriber
        return None

    async def async_send_events(
        self, subscriber: Optional[EventSubscriber] = None
    ) -> None:
        """Send event updates to any subscribers."""
        if not subscriber:
            now = datetime.now()
            self._subscribers = [
                _sub for _sub in self._subscribers if now < _sub.expiration
            ]
            subscribers = self._subscribers
            if not self._subscribers:
                return
        else:
            subscribers = [subscriber]
        event_el = ET.Element("e:propertyset")
        event_el.set("xmlns:e", "urn:schemas-upnp-org:event-1-0")
        for state_var in self.state_variables.values():
            if not isinstance(state_var, UpnpEventableStateVariable):
                continue
            prop_el = ET.SubElement(event_el, "e:property")
            ET.SubElement(prop_el, state_var.name).text = str(state_var.value)
        message = ET.tostring(event_el, encoding="utf-8", xml_declaration=True).decode()

        headers = {
            "CONTENT-TYPE": 'text/xml; charset="utf-8"',
            "NT": "upnp:event",
            "NTS": "upnp:propchange",
        }
        tasks = []
        for sub in subscribers:
            hdr = headers.copy()
            hdr["SID"] = sub.uuid
            hdr["SEQ"] = str(sub.get_next_seq())
            tasks.append(
                self.requester.async_http_request(
                    HttpRequest("NOTIFY", sub.url, headers=hdr, body=message)
                )
            )
        await asyncio.gather(*tasks)


class UpnpServerDevice(UpnpDevice):
    """UPnP Device representation."""

    DEVICE_DEFINITION: DeviceInfo
    EMBEDDED_DEVICES: Sequence[Type["UpnpServerDevice"]]
    SERVICES: Sequence[Type[UpnpServerService]]
    ROUTES: Optional[Sequence[RouteDef]] = None

    def __init__(
        self,
        requester: UpnpRequester,
        base_uri: str,
        boot_id: int = 1,
        config_id: int = 1,
    ) -> None:
        """Initialize."""
        services = [service_type(requester=requester) for service_type in self.SERVICES]
        embedded_devices = [
            device_type(
                requester=requester,
                base_uri=base_uri,
                boot_id=boot_id,
                config_id=config_id,
            )
            for device_type in self.EMBEDDED_DEVICES
        ]
        super().__init__(
            requester=requester,
            device_info=self.DEVICE_DEFINITION,
            services=services,
            embedded_devices=embedded_devices,
        )
        self.base_uri = base_uri
        self.host = urlparse(base_uri).hostname
        self.boot_id = boot_id
        self.config_id = config_id


class SsdpSearchResponder:
    """SSDP SEARCH responder."""

    def __init__(
        self,
        device: UpnpServerDevice,
        source: Optional[AddressTupleVXType] = None,
        target: Optional[AddressTupleVXType] = None,
        options: Optional[Dict[str, Any]] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        """Init the ssdp search responder class."""
        # pylint: disable=too-many-arguments,too-many-positional-arguments
        self.device = device
        self.source, self.target = determine_source_target(source, target)
        self.options = options or {}

        self._transport: Optional[DatagramTransport] = None
        self._response_transport: Optional[DatagramTransport] = None
        self._loop = loop or asyncio.get_running_loop()

    def _on_connect_response(self, transport: DatagramTransport) -> None:
        """Handle on connect for response."""
        _LOGGER.debug("Connected to response transport: %s", transport)
        self._response_transport = transport

    def _on_connect(self, transport: DatagramTransport) -> None:
        """Handle on connect."""
        self._transport = transport

    def _on_data(
        self,
        request_line: str,
        headers: CaseInsensitiveDict,
    ) -> None:
        """Handle data."""
        # pylint: disable=too-many-branches
        assert self._transport

        if (
            request_line != "M-SEARCH * HTTP/1.1"
            or headers.get_lower("man") != SSDP_DISCOVER
        ):
            return

        remote_addr = cast(AddressTupleVXType, headers.get_lower("_remote_addr"))
        debug = _LOGGER.isEnabledFor(logging.DEBUG)
        if debug:  # pragma: no branch
            _LOGGER.debug(
                "Received M-SEARCH from: %s, headers: %s", remote_addr, headers
            )

        mx_header = headers.get_lower("mx")
        delay = 0
        if mx_header is not None:
            try:
                delay = min(5, int(mx_header))
                if debug:  # pragma: no branch
                    _LOGGER.debug("Deferring response for %d seconds", delay)
            except ValueError:
                pass

        if not (responses := self._build_responses(headers)):
            return

        if delay:
            # The delay should be random between 0 and MX.
            # We use between 0.100 and MX-0.250 seconds to avoid
            # flooding the network with simultaneous responses.
            #
            # We do not set the upper limit to exactly MX seconds
            # because it might take up to 0.250 seconds to send the
            # response, and we want to avoid sending the response
            # after the MX timeout.
            self._loop.call_at(
                self._loop.time() + randrange(100, (delay * 1000) - 250) / 1000,
                self._send_responses,
                remote_addr,
                responses,
            )
        self._send_responses(remote_addr, responses)

    def _build_responses(self, headers: CaseInsensitiveDict) -> List[bytes]:
        # Determine how we should respond, page 1.3.2 of UPnP-arch-DeviceArchitecture-v2.0.
        st_header: str = headers.get_lower("st", "")
        search_target = st_header.lower()
        responses: List[bytes] = []

        if search_target == SSDP_ST_ALL:
            # 3 + 2d + k (d: embedded device, k: service)
            # global:      ST: upnp:rootdevice
            #              USN: uuid:device-UUID::upnp:rootdevice
            # per device : ST: uuid:device-UUID
            #              USN: uuid:device-UUID
            # per device : ST: urn:schemas-upnp-org:device:deviceType:ver
            #              USN: uuid:device-UUID::urn:schemas-upnp-org:device:deviceType:ver
            # per service: ST: urn:schemas-upnp-org:service:serviceType:ver
            #              USN: uuid:device-UUID::urn:schemas-upnp-org:service:serviceType:ver
            all_devices = self.device.all_devices
            all_services = self.device.all_services
            responses.append(self._build_response_rootdevice())
            responses.extend(
                self._build_responses_device_udn(device) for device in all_devices
            )
            responses.extend(
                self._build_responses_device_type(device) for device in all_devices
            )
            responses.extend(
                self._build_responses_service(service) for service in all_services
            )
        elif search_target == SSDP_ST_ROOTDEVICE:
            responses.append(self._build_response_rootdevice())
        elif matched_devices := self.device.get_devices_matching_udn(search_target):
            responses.extend(
                self._build_responses_device_udn(device) for device in matched_devices
            )
        elif matched_devices := self._matched_devices_by_type(search_target):
            responses.extend(
                self._build_responses_device_type(device, search_target)
                for device in matched_devices
            )
        elif matched_services := self._matched_services_by_type(search_target):
            responses.extend(
                self._build_responses_service(service, search_target)
                for service in matched_services
            )

        if self.options.get(SSDP_SEARCH_RESPONDER_OPTION_ALWAYS_REPLY_WITH_ROOT_DEVICE):
            responses.append(self._build_response_rootdevice())

        return responses

    @staticmethod
    def _match_type_versions(type_ver: str, search_target: str) -> bool:
        """Determine if any service/device type up to the max version matches search_target."""
        # As per 1.3.2 of the UPnP Device Architecture spec, all device service types
        # must respond to and be backwards-compatible with older versions of the same type
        type_ver_lower: str = type_ver.lower()
        try:
            base, max_ver = type_ver_lower.rsplit(":", 1)
            max_ver_i = int(max_ver)
            for ver in range(max_ver_i + 1):
                if f"{base}:{ver}" == search_target:
                    return True
        except ValueError:
            if type_ver_lower == search_target:
                return True
        return False

    def _matched_devices_by_type(self, search_target: str) -> List[UpnpDevice]:
        """Get matched devices by device type."""
        return [
            device
            for device in self.device.all_devices
            if self._match_type_versions(device.device_type, search_target)
        ]

    def _matched_services_by_type(self, search_target: str) -> List[UpnpService]:
        """Get matched services by service type."""
        return [
            service
            for service in self.device.all_services
            if self._match_type_versions(service.service_type, search_target)
        ]

    async def async_start(self) -> None:
        """Start."""
        _LOGGER.debug("Start listening for search requests")

        # Create response socket/protocol.
        response_sock, _source, _target = get_ssdp_socket(self.source, self.target)

        await self._loop.create_datagram_endpoint(
            lambda: SsdpProtocol(
                self._loop,
                on_connect=self._on_connect_response,
            ),
            sock=response_sock,
        )

        # Create listening socket/protocol.
        sock, _source, _target = get_ssdp_socket(self.source, self.target)

        address = ("", self.target[1])
        _LOGGER.debug("Binding socket, socket: %s, address: %s", sock, address)
        sock.bind(address)

        await self._loop.create_datagram_endpoint(
            lambda: SsdpProtocol(
                self._loop,
                on_connect=self._on_connect,
                on_data=self._on_data,
            ),
            sock=sock,
        )

    async def async_stop(self) -> None:
        """Stop listening for advertisements."""
        assert self._transport

        _LOGGER.debug("Stop listening for SEARCH requests")
        self._transport.close()

    def _build_response_rootdevice(self) -> bytes:
        """Send root device response."""
        return self._build_response(
            "upnp:rootdevice", f"{self.device.udn}::upnp:rootdevice"
        )

    def _build_responses_device_udn(self, device: UpnpDevice) -> bytes:
        """Send device responses for UDN."""
        return self._build_response(device.udn, f"{self.device.udn}")

    def _build_responses_device_type(
        self, device: UpnpDevice, device_type: Optional[str] = None
    ) -> bytes:
        """Send device responses for device type."""
        return self._build_response(
            device_type or device.device_type,
            f"{self.device.udn}::{device.device_type}",
        )

    def _build_responses_service(
        self, service: UpnpService, service_type: Optional[str] = None
    ) -> bytes:
        """Send service responses."""
        return self._build_response(
            service_type or service.service_type,
            f"{self.device.udn}::{service.service_type}",
        )

    def _build_response(
        self,
        service_type: str,
        unique_service_name: str,
    ) -> bytes:
        """Send a response."""
        return build_ssdp_packet(
            "HTTP/1.1 200 OK",
            {
                "CACHE-CONTROL": HEADER_CACHE_CONTROL,
                "DATE": format_date_time(time.time()),
                "SERVER": HEADER_SERVER,
                "ST": service_type,
                "USN": unique_service_name,
                "EXT": "",
                "LOCATION": f"{self.device.base_uri}{self.device.device_url}",
                "BOOTID.UPNP.ORG": str(self.device.boot_id),
                "CONFIGID.UPNP.ORG": str(self.device.config_id),
            },
        )

    def _send_responses(
        self, remote_addr: AddressTupleVXType, responses: List[bytes]
    ) -> None:
        """Send responses."""
        assert self._response_transport
        if _LOGGER.isEnabledFor(logging.DEBUG):  # pragma: no branch
            sock: Optional[socket.socket] = self._response_transport.get_extra_info(
                "socket"
            )
            _LOGGER.debug(
                "Sending SSDP packet, transport: %s, socket: %s, target: %s",
                self._response_transport,
                sock,
                remote_addr,
            )
        _LOGGER_TRAFFIC_SSDP.debug(
            "Sending SSDP packets, target: %s, data: %s", remote_addr, responses
        )
        for response in responses:
            try:
                protocol = cast(SsdpProtocol, self._response_transport.get_protocol())
                protocol.send_ssdp_packet(response, remote_addr)
            except OSError as err:
                _LOGGER.debug("Error sending response: %s", err)


def _build_advertisements(
    target: AddressTupleVXType,
    root_device: UpnpServerDevice,
    nts: NotificationSubType = NotificationSubType.SSDP_ALIVE,
) -> List[CaseInsensitiveDict]:
    """Build advertisements to be sent for a UpnpDevice."""
    # 3 + 2d + k (d: embedded device, k: service)
    # global:      ST: upnp:rootdevice
    #              USN: uuid:device-UUID::upnp:rootdevice
    # per device : ST: uuid:device-UUID
    #              USN: uuid:device-UUID
    # per device : ST: urn:schemas-upnp-org:device:deviceType:ver
    #              USN: uuid:device-UUID::urn:schemas-upnp-org:device:deviceType:ver
    # per service: ST: urn:schemas-upnp-org:service:serviceType:ver
    #              USN: uuid:device-UUID::urn:schemas-upnp-org:service:serviceType:ver
    advertisements: List[CaseInsensitiveDict] = []

    host = (
        f"[{target[0]}]:{target[1]}"
        if is_ipv6_address(target)
        else f"{target[0]}:{target[1]}"
    )
    base_headers = {
        "NTS": nts.value,
        "HOST": host,
        "CACHE-CONTROL": HEADER_CACHE_CONTROL,
        "SERVER": HEADER_SERVER,
        "BOOTID.UPNP.ORG": str(root_device.boot_id),
        "CONFIGID.UPNP.ORG": str(root_device.config_id),
        "LOCATION": f"{root_device.base_uri}{root_device.device_url}",
    }

    # root device
    advertisements.append(
        CaseInsensitiveDict(
            base_headers,
            NT="upnp:rootdevice",
            USN=f"{root_device.udn}::upnp:rootdevice",
        )
    )

    for device in root_device.all_devices:
        advertisements.append(
            CaseInsensitiveDict(
                base_headers,
                NT=f"{device.udn}",
                USN=f"{device.udn}",
            )
        )
        advertisements.append(
            CaseInsensitiveDict(
                base_headers,
                NT=f"{device.device_type}",
                USN=f"{device.udn}::{device.device_type}",
            )
        )

    for service in root_device.all_services:
        advertisements.append(
            CaseInsensitiveDict(
                base_headers,
                NT=f"{service.service_type}",
                USN=f"{service.device.udn}::{service.service_type}",
            )
        )

    return advertisements


class SsdpAdvertisementAnnouncer:
    """SSDP Advertisement announcer."""

    # pylint: disable=too-many-instance-attributes

    ANNOUNCE_INTERVAL = timedelta(seconds=30)

    def __init__(
        self,
        device: UpnpServerDevice,
        source: Optional[AddressTupleVXType] = None,
        target: Optional[AddressTupleVXType] = None,
        options: Optional[Dict[str, Any]] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        """Init the ssdp search responder class."""
        # pylint: disable=too-many-arguments,too-many-positional-arguments
        self.device = device
        self.source, self.target = determine_source_target(source, target)
        self.options = options or {}
        self._loop = loop or asyncio.get_running_loop()

        self._transport: Optional[DatagramTransport] = None
        advertisements = _build_advertisements(self.target, device)
        self._advertisements = cycle(advertisements)
        self._cancel_announce: Optional[asyncio.TimerHandle] = None

    def _on_connect(self, transport: DatagramTransport) -> None:
        """Handle on connect."""
        self._transport = transport

    async def async_start(self) -> None:
        """Start."""
        _LOGGER.debug("Start advertisements announcer")

        # Construct a socket for use with this pairs of endpoints.
        sock, _source, _target = get_ssdp_socket(self.source, self.target)
        if sys.platform.startswith("win32"):
            address = self.source
            _LOGGER.debug("Binding socket, socket: %s, address: %s", sock, address)
            sock.bind(address)

        # Create protocol and send discovery packet.
        await self._loop.create_datagram_endpoint(
            lambda: SsdpProtocol(
                self._loop,
                on_connect=self._on_connect,
            ),
            sock=sock,
        )

        await self.async_wait_for_transport_protocol()

        # Announce and reschedule self.
        self._announce_next()

    async def async_stop(self) -> None:
        """Stop listening for advertisements."""
        assert self._transport

        sock: Optional[socket.socket] = self._transport.get_extra_info("socket")
        _LOGGER.debug(
            "Stop advertisements announcer, transport: %s, socket: %s",
            self._transport,
            sock,
        )

        if self._cancel_announce is not None:
            self._cancel_announce.cancel()

        self._send_byebyes()
        self._transport.close()

    async def async_wait_for_transport_protocol(self) -> None:
        """Wait for the protocol to become available."""
        for _ in range(0, 5):
            if (
                self._transport is not None
                and self._transport.get_protocol() is not None
            ):
                break

            await asyncio.sleep(0.1)
        else:
            raise UpnpError("Failed to get protocol")

    def _announce_next(self) -> None:
        """Announce next advertisement."""
        _LOGGER.debug("Announcing")
        assert self._transport

        protocol = cast(SsdpProtocol, self._transport.get_protocol())
        start_line = "NOTIFY * HTTP/1.1"
        headers = next(self._advertisements)
        packet = build_ssdp_packet(start_line, headers)

        _LOGGER.debug(
            "Sending advertisement, NTS: %s, NT: %s, USN: %s",
            headers["NTS"],
            headers["NT"],
            headers["USN"],
        )
        protocol.send_ssdp_packet(packet, self.target)

        # Reschedule self.
        self._cancel_announce = self._loop.call_later(
            SsdpAdvertisementAnnouncer.ANNOUNCE_INTERVAL.total_seconds(),
            self._announce_next,
        )

    def _send_byebyes(self) -> None:
        """Send ssdp:byebye."""
        assert self._transport

        start_line = "NOTIFY * HTTP/1.1"
        advertisements = _build_advertisements(
            self.target, self.device, NotificationSubType.SSDP_BYEBYE
        )
        for headers in advertisements:
            packet = build_ssdp_packet(start_line, headers)
            protocol = cast(SsdpProtocol, self._transport.get_protocol())
            _LOGGER.debug(
                "Sending advertisement, NTS: %s, NT: %s, USN: %s",
                headers["NTS"],
                headers["NT"],
                headers["USN"],
            )
            protocol.send_ssdp_packet(packet, self.target)


class UpnpXmlSerializer:
    """Helper class to create device/service description from UpnpDevice/UpnpService."""

    # pylint: disable=too-few-public-methods

    @classmethod
    def to_xml(cls, thing: Union[UpnpDevice, UpnpService]) -> ET.Element:
        """Convert thing to XML."""
        if isinstance(thing, UpnpDevice):
            return cls._device_to_xml(thing)
        if isinstance(thing, UpnpService):
            return cls._service_to_xml(thing)

        raise NotImplementedError()

    @classmethod
    def _device_to_xml(cls, device: UpnpDevice) -> ET.Element:
        """Convert device to device description XML."""
        root_el = ET.Element("root", xmlns="urn:schemas-upnp-org:device-1-0")
        spec_version_el = ET.SubElement(root_el, "specVersion")
        ET.SubElement(spec_version_el, "major").text = "1"
        ET.SubElement(spec_version_el, "minor").text = "0"

        device_el = cls._device_to_xml_bare(device)
        root_el.append(device_el)

        return root_el

    @classmethod
    def _device_to_xml_bare(cls, device: UpnpDevice) -> ET.Element:
        """Convert device to XML, without the root-element."""
        device_el = ET.Element("device", xmlns="urn:schemas-upnp-org:device-1-0")
        ET.SubElement(device_el, "deviceType").text = device.device_type
        ET.SubElement(device_el, "friendlyName").text = device.friendly_name
        ET.SubElement(device_el, "manufacturer").text = device.manufacturer
        ET.SubElement(device_el, "manufacturerURL").text = device.manufacturer_url
        ET.SubElement(device_el, "modelDescription").text = device.model_description
        ET.SubElement(device_el, "modelName").text = device.model_name
        ET.SubElement(device_el, "modelNumber").text = device.model_number
        ET.SubElement(device_el, "modelURL").text = device.model_url
        ET.SubElement(device_el, "serialNumber").text = device.serial_number
        ET.SubElement(device_el, "UDN").text = device.udn
        ET.SubElement(device_el, "UPC").text = device.upc
        ET.SubElement(device_el, "presentationURL").text = device.presentation_url

        icon_list_el = ET.SubElement(device_el, "iconList")
        for icon in device.icons:
            icon_el = ET.SubElement(icon_list_el, "icon")
            ET.SubElement(icon_el, "mimetype").text = icon.mimetype
            ET.SubElement(icon_el, "width").text = str(icon.width)
            ET.SubElement(icon_el, "height").text = str(icon.height)
            ET.SubElement(icon_el, "depth").text = str(icon.depth)
            ET.SubElement(icon_el, "url").text = icon.url

        service_list_el = ET.SubElement(device_el, "serviceList")
        for service in device.services.values():
            service_el = ET.SubElement(service_list_el, "service")
            ET.SubElement(service_el, "serviceType").text = service.service_type
            ET.SubElement(service_el, "serviceId").text = service.service_id
            ET.SubElement(service_el, "controlURL").text = service.control_url
            ET.SubElement(service_el, "eventSubURL").text = service.event_sub_url
            ET.SubElement(service_el, "SCPDURL").text = service.scpd_url

        device_list_el = ET.SubElement(device_el, "deviceList")
        for embedded_device in device.embedded_devices.values():
            embedded_device_el = cls._device_to_xml_bare(embedded_device)
            device_list_el.append(embedded_device_el)

        return device_el

    @classmethod
    def _service_to_xml(cls, service: UpnpService) -> ET.Element:
        """Convert service to service description XML."""
        scpd_el = ET.Element("scpd", xmlns="urn:schemas-upnp-org:service-1-0")
        spec_version_el = ET.SubElement(scpd_el, "specVersion")
        ET.SubElement(spec_version_el, "major").text = "1"
        ET.SubElement(spec_version_el, "minor").text = "0"

        action_list_el = ET.SubElement(scpd_el, "actionList")
        for action in service.actions.values():
            action_el = cls._action_to_xml(action)
            action_list_el.append(action_el)

        state_table_el = ET.SubElement(scpd_el, "serviceStateTable")
        for state_var in service.state_variables.values():
            state_var_el = cls._state_variable_to_xml(state_var)
            state_table_el.append(state_var_el)

        return scpd_el

    @classmethod
    def _action_to_xml(cls, action: UpnpAction) -> ET.Element:
        """Convert action to service description XML."""
        action_el = ET.Element("action")
        ET.SubElement(action_el, "name").text = action.name

        if action.arguments:
            arg_list_el = ET.SubElement(action_el, "argumentList")
            for arg in action.in_arguments():
                arg_el = cls._action_argument_to_xml(arg)
                arg_list_el.append(arg_el)
            for arg in action.out_arguments():
                arg_el = cls._action_argument_to_xml(arg)
                arg_list_el.append(arg_el)

        return action_el

    @classmethod
    def _action_argument_to_xml(cls, argument: UpnpAction.Argument) -> ET.Element:
        """Convert action argument to service description XML."""
        arg_el = ET.Element("argument")
        ET.SubElement(arg_el, "name").text = argument.name
        ET.SubElement(arg_el, "direction").text = argument.direction
        ET.SubElement(arg_el, "relatedStateVariable").text = (
            argument.related_state_variable.name
        )
        return arg_el

    @classmethod
    def _state_variable_to_xml(cls, state_variable: UpnpStateVariable) -> ET.Element:
        """Convert state variable to service description XML."""
        state_var_el = ET.Element(
            "stateVariable", sendEvents="yes" if state_variable.send_events else "no"
        )
        ET.SubElement(state_var_el, "name").text = state_variable.name
        ET.SubElement(state_var_el, "dataType").text = state_variable.data_type

        if state_variable.allowed_values:
            value_list_el = ET.SubElement(state_var_el, "allowedValueList")
            for allowed_value in state_variable.allowed_values:
                ET.SubElement(value_list_el, "allowedValue").text = str(allowed_value)

        if None not in (state_variable.min_value, state_variable.max_value):
            value_range_el = ET.SubElement(state_var_el, "allowedValueRange")
            ET.SubElement(value_range_el, "minimum").text = str(
                state_variable.min_value
            )
            ET.SubElement(value_range_el, "maximum").text = str(
                state_variable.max_value
            )

        if state_variable.default_value is not None:
            ET.SubElement(state_var_el, "defaultValue").text = str(
                state_variable.default_value
            )

        return state_var_el


def callable_action(
    name: str, in_args: Mapping[str, str], out_args: Mapping[str, str]
) -> Callable:
    """Declare method as a callable UpnpAction."""

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)

        setattr(wrapper, "__upnp_action__", [name, in_args, out_args])

        return wrapper

    return decorator


async def _parse_action_body(
    service: UpnpServerService, request: Request
) -> Tuple[str, Dict[str, Any]]:
    """Parse action body."""
    # Parse call.
    soap_action = request.headers.get("SOAPAction", "").strip('"')
    try:
        _, action_name = soap_action.split("#")
        data = await request.text()
        root_el: ET.Element = DET.fromstring(data)
        body_el = root_el.find("s:Body", NAMESPACES)
        assert body_el
        rpc_el = body_el[0]
    except Exception as exc:
        raise HTTPBadRequest(reason="InvalidSoap") from exc

    if action_name not in service.actions:
        raise HTTPBadRequest(reason="InvalidAction")

    kwargs: Dict[str, Any] = {}
    action = service.action(action_name)
    for arg in rpc_el:
        action_arg = action.argument(arg.tag, direction="in")
        if action_arg is None:
            raise HTTPBadRequest(reason="InvalidActionArgument")
        state_var = action_arg.related_state_variable
        kwargs[arg.tag] = state_var.coerce_python(arg.text or "")

    return action_name, kwargs


def _create_action_response(
    service: UpnpServerService, action_name: str, result: Dict[str, Any]
) -> Response:
    """Create action call response."""
    envelope_el = ET.Element(
        "s:Envelope",
        attrib={
            "xmlns:s": NAMESPACES["s"],
            "s:encodingStyle": NAMESPACES["es"],
        },
    )
    body_el = ET.SubElement(envelope_el, "s:Body")

    response_el = ET.SubElement(
        body_el, f"st:{action_name}Response", attrib={"xmlns:st": service.service_type}
    )
    out_state_vars = {
        var.name: var.related_state_variable
        for var in service.actions[action_name].out_arguments()
    }
    for key, value in result.items():
        if isinstance(value, UpnpStateVariable):
            ET.SubElement(response_el, key).text = value.upnp_value
        else:
            template_var = out_state_vars[key]
            template_var.validate_value(value)
            ET.SubElement(response_el, key).text = template_var.coerce_upnp(value)
    return Response(
        content_type="text/xml",
        charset="utf-8",
        body=ET.tostring(envelope_el, encoding="utf-8", xml_declaration=True),
    )


def _create_error_action_response(
    exception: UpnpError,
) -> Response:
    """Create action call response."""
    envelope_el = ET.Element(
        "s:Envelope",
        attrib={
            "xmlns:s": NAMESPACES["s"],
            "s:encodingStyle": NAMESPACES["es"],
        },
    )
    body_el = ET.SubElement(envelope_el, "s:Body")
    fault_el = ET.SubElement(body_el, "s:Fault")
    ET.SubElement(fault_el, "faultcode").text = "s:Client"
    ET.SubElement(fault_el, "faultstring").text = "UPnPError"
    detail_el = ET.SubElement(fault_el, "detail")
    error_el = ET.SubElement(
        detail_el, "UPnPError", xmlns="urn:schemas-upnp-org:control-1-0"
    )
    error_code = (
        exception.error_code or UpnpActionErrorCode.ACTION_FAILED.value
        if isinstance(exception, UpnpActionError)
        else 402 if isinstance(exception, UpnpValueError) else 501
    )
    ET.SubElement(error_el, "errorCode").text = str(error_code)
    ET.SubElement(error_el, "errorDescription").text = "Action Failed"

    return Response(
        status=500,
        content_type="text/xml",
        charset="utf-8",
        body=ET.tostring(envelope_el, encoding="utf-8", xml_declaration=True),
    )


async def action_handler(service: UpnpServerService, request: Request) -> Response:
    """Handle action."""
    action_name, kwargs = await _parse_action_body(service, request)

    # Do call.
    try:
        call_result = await service.async_handle_action(action_name, **kwargs)
    except UpnpValueError as exc:
        return _create_error_action_response(exc)
    except UpnpActionError as exc:
        return _create_error_action_response(exc)

    return _create_action_response(service, action_name, call_result)


async def subscribe_handler(service: UpnpServerService, request: Request) -> Response:
    """SUBSCRIBE handler."""
    callback_url = request.headers.get("CALLBACK", None)
    timeout = request.headers.get("TIMEOUT", None)
    sid = request.headers.get("SID", None)

    timeout_val = None
    if timeout is not None:
        try:
            timeout_val = int(timeout.lower().replace("second-", ""))
        except ValueError:
            return Response(status=400)

    subscriber = None
    if sid:
        subscriber = service.get_subscriber(sid)
        if subscriber:
            subscriber.timeout = timeout_val
    else:
        if callback_url:
            # callback url is specified as <http://...>
            # remove the outside <>
            callback_url = callback_url[1:-1]
            subscriber = EventSubscriber(callback_url, timeout_val)

    if not subscriber:
        return Response(status=404)

    headers = {
        "DATE": format_date_time(mktime(datetime.now().timetuple())),
        "SERVER": HEADER_SERVER,
        "SID": subscriber.uuid,
        "TIMEOUT": str(subscriber.timeout),
    }
    resp = Response(status=200, headers=headers)
    if sid is None:
        # this is an initial subscription.  Need to send state-vars
        # AFTER response completion
        await resp.prepare(request)
        await resp.write_eof()
        await service.async_send_events(subscriber)
        service.add_subscriber(subscriber)
    return resp


async def unsubscribe_handler(service: UpnpServerService, request: Request) -> Response:
    """UNSUBSCRIBE handler."""
    sid = request.headers.get("SID", None)
    if sid:
        if service.del_subscriber(sid):
            return Response(status=200)
    return Response(status=412)


async def to_xml(
    thing: Union[UpnpServerDevice, UpnpServerService], _request: Request
) -> Response:
    """Construct device/service description."""
    serializer = UpnpXmlSerializer()
    thing_el = serializer.to_xml(thing)
    encoding = "utf-8"
    thing_xml = ET.tostring(thing_el, encoding=encoding, xml_declaration=True)
    return Response(content_type="text/xml", charset=encoding, body=thing_xml)


def create_state_var(
    data_type: str,
    *,
    allowed: Optional[List[str]] = None,
    allowed_range: Optional[Mapping[str, Optional[str]]] = None,
    default: Optional[str] = None,
) -> StateVariableTypeInfo:
    """Create state variables."""
    return StateVariableTypeInfo(
        data_type=data_type,
        data_type_mapping=STATE_VARIABLE_TYPE_MAPPING[data_type],
        default_value=default,
        allowed_value_range=allowed_range or {},
        allowed_values=allowed,
        xml=ET.Element("server_stateVariable"),
    )


def create_event_var(
    data_type: str,
    *,
    allowed: Optional[List[str]] = None,
    allowed_range: Optional[Mapping[str, Optional[str]]] = None,
    default: Optional[str] = None,
    max_rate: Optional[float] = None,
) -> StateVariableTypeInfo:
    """Create event variables."""
    return cast(
        StateVariableTypeInfo,
        EventableStateVariableTypeInfo(
            data_type=data_type,
            data_type_mapping=STATE_VARIABLE_TYPE_MAPPING[data_type],
            default_value=default,
            allowed_value_range=allowed_range or {},
            allowed_values=allowed,
            max_rate=max_rate,
            xml=ET.Element("server_stateVariable"),
        ),
    )


class UpnpServer:
    """UPnP Server."""

    # pylint: disable=too-many-instance-attributes

    def __init__(
        self,
        server_device: Type[UpnpServerDevice],
        source: AddressTupleVXType,
        target: Optional[AddressTupleVXType] = None,
        http_port: Optional[int] = None,
        boot_id: int = 1,
        config_id: int = 1,
        options: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Initialize."""
        # pylint: disable=too-many-arguments,too-many-positional-arguments
        self.server_device = server_device
        self.source, self.target = determine_source_target(source, target)
        self.http_port = http_port
        self.boot_id = boot_id
        self.config_id = config_id
        self.options = options or {}

        self.base_uri: Optional[str] = None
        self._device: Optional[UpnpServerDevice] = None
        self._site: Optional[TCPSite] = None
        self._search_responder: Optional[SsdpSearchResponder] = None
        self._advertisement_announcer: Optional[SsdpAdvertisementAnnouncer] = None

    async def async_start(self) -> None:
        """Start."""
        self._create_device()
        await self._async_start_http_server()
        await self._async_start_ssdp()

    def _create_device(self) -> None:
        """Create device."""
        requester = AiohttpRequester()
        is_ipv6 = ":" in self.source[0]
        self.base_uri = (
            f"http://[{self.source[0]}]:{self.http_port}"
            if is_ipv6
            else f"http://{self.source[0]}:{self.http_port}"
        )
        self._device = self.server_device(
            requester, self.base_uri, self.boot_id, self.config_id
        )

    async def _async_start_http_server(self) -> None:
        """Start http server."""
        assert self._device

        # Build app.
        app = Application()
        app.router.add_get(self._device.device_url, partial(to_xml, self._device))

        for service in self._device.all_services:
            service = cast(UpnpServerService, service)
            app.router.add_get(
                service.SERVICE_DEFINITION.scpd_url, partial(to_xml, service)
            )
            app.router.add_post(
                service.SERVICE_DEFINITION.control_url, partial(action_handler, service)
            )
            app.router.add_route(
                "SUBSCRIBE",
                service.SERVICE_DEFINITION.event_sub_url,
                partial(subscribe_handler, service),
            )
            app.router.add_route(
                "UNSUBSCRIBE",
                service.SERVICE_DEFINITION.event_sub_url,
                partial(unsubscribe_handler, service),
            )

        if self._device.ROUTES:
            app.router.add_routes(self._device.ROUTES)

        # Create AppRunner.
        runner = AppRunner(app, access_log=_LOGGER_TRAFFIC_UPNP)
        await runner.setup()

        # Launch TCP handler.
        is_ipv6 = ":" in self.source[0]
        host = f"{self.source[0]}%{self.source[3]}" if is_ipv6 else self.source[0]  # type: ignore
        self._site = TCPSite(runner, host, self.http_port, reuse_address=True)
        await self._site.start()

        assert self._device
        _LOGGER.debug(
            "Device listening at %s%s", self._site.name, self._device.device_url
        )

    async def _async_start_ssdp(self) -> None:
        """Start SSDP handling."""
        _LOGGER.debug(
            "Starting SSDP handling, source: %s, target: %s", self.source, self.target
        )
        assert self._device
        self._search_responder = SsdpSearchResponder(
            self._device,
            source=self.source,
            target=self.target,
            options=self.options.get(SSDP_SEARCH_RESPONDER_OPTIONS),
        )
        self._advertisement_announcer = SsdpAdvertisementAnnouncer(
            self._device,
            source=self.source,
            target=self.target,
            options=self.options.get(SSDP_ADVERTISEMENT_ANNOUNCER_OPTIONS),
        )

        await self._search_responder.async_start()
        await self._advertisement_announcer.async_start()

    async def async_stop(self) -> None:
        """Stop server."""
        await self._async_stop_ssdp()
        await self._async_stop_http_server()

    async def _async_stop_ssdp(self) -> None:
        """Stop SSDP handling."""
        if self._advertisement_announcer:
            await self._advertisement_announcer.async_stop()
        if self._search_responder:
            await self._search_responder.async_stop()

    async def _async_stop_http_server(self) -> None:
        """Stop HTTP server."""
        if self._site:
            await self._site.stop()