api.py

"""
Implements an asynchronous interface for a Frontier Silicon device.

For example internet radios from: Medion, Hama, Auna, ...
"""

import asyncio
from asyncio.exceptions import TimeoutError
import typing as t
import logging
from afsapi.exceptions import (
    FSApiException,
    InvalidPinException,
    InvalidSessionException,
    NotImplementedException,
    OutOfRangeException,
    ConnectionError,
)
from afsapi.models import Preset, Equaliser, PlayerMode, PlayControl, PlayState
from afsapi.throttler import Throttler
from afsapi.utils import unpack_xml, maybe
from enum import Enum
import aiohttp
import xml.etree.ElementTree as ET

DataItem = t.Union[str, int]


DEFAULT_TIMEOUT_IN_SECONDS = 15

TIME_AFTER_READ_CALLS_IN_SECONDS = 0
TIME_AFTER_SET_CALLS_IN_SECONDS = 0.3
TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS = 1.0

FSApiValueType = Enum("FSApiValueType", "TEXT BOOL INT LONG SIGNED_LONG")

VALUE_TYPE_TO_XML_PATH = {
    FSApiValueType.TEXT: "c8_array",
    FSApiValueType.INT: "u8",
    FSApiValueType.LONG: "u32",
    FSApiValueType.SIGNED_LONG: "s32",
}

READ_ONLY = False
READ_WRITE = True

# implemented API calls
API = {
    # sys
    "power": "netRemote.sys.power",
    "mode": "netRemote.sys.mode",
    "wired_mac": "netRemote.sys.net.wired.macAddress",
    "wired_active": "netRemote.sys.net.wired.interfaceEnable",
    "wlan_mac": "netRemote.sys.net.wlan.macAddress",
    "wlan_active": "netRemote.sys.net.wlan.interfaceEnable",
    "rssi": "netRemote.sys.net.wlan.rssi",
    # sys.info
    "friendly_name": "netRemote.sys.info.friendlyName",
    "radio_id": "netRemote.sys.info.radioId",
    "version": "netRemote.sys.info.version",
    # sys.caps
    "valid_modes": "netRemote.sys.caps.validModes",
    "equalisers": "netRemote.sys.caps.eqPresets",
    "sleep": "netRemote.sys.sleep",
    # sys.audio
    "eqpreset": "netRemote.sys.audio.eqpreset",
    "eqloudness": "netRemote.sys.audio.eqloudness",
    "bass": "netRemote.sys.audio.eqcustom.param0",
    "treble": "netRemote.sys.audio.eqcustom.param1",
    # volume
    "volume_steps": "netRemote.sys.caps.volumeSteps",
    "volume": "netRemote.sys.audio.volume",
    "mute": "netRemote.sys.audio.mute",
    # play
    "status": "netRemote.play.status",
    "name": "netRemote.play.info.name",
    "control": "netRemote.play.control",
    "shuffle": "netRemote.play.shuffle",
    "repeat": "netRemote.play.repeat",
    "position": "netRemote.play.position",
    "rate": "netRemote.play.rate",
    # info
    "text": "netRemote.play.info.text",
    "artist": "netRemote.play.info.artist",
    "album": "netRemote.play.info.album",
    "graphic_uri": "netRemote.play.info.graphicUri",
    "duration": "netRemote.play.info.duration",
    # nav
    "nav_state": "netRemote.nav.state",
    "numitems": "netRemote.nav.numitems",
    "nav_list": "netRemote.nav.list",
    "navigate": "netRemote.nav.action.navigate",
    "selectItem": "netRemote.nav.action.selectItem",
    "presets": "netRemote.nav.presets",
    "selectPreset": "netRemote.nav.action.selectPreset",
}

LOGGER = logging.getLogger(__name__)

# pylint: disable=R0904


class AFSAPI:
    """Builds the interface to a Frontier Silicon device."""

    def __init__(
        self,
        webfsapi_endpoint: str,
        pin: t.Union[str, int],
        timeout: int = DEFAULT_TIMEOUT_IN_SECONDS,
    ):
        """Initialize the Frontier Silicon device."""
        self.webfsapi_endpoint = webfsapi_endpoint
        self.pin = str(pin)
        self.timeout = timeout

        self.sid: t.Optional[str] = None
        self.__volume_steps: t.Optional[int] = None

        self.__modes = None
        self.__equalisers = None

        self._current_nav_path: list[int] = []

        self.__throttler = Throttler()

    @staticmethod
    async def get_webfsapi_endpoint(
        fsapi_device_url: str, timeout: int = DEFAULT_TIMEOUT_IN_SECONDS
    ) -> str:

        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(force_close=True),
            timeout=aiohttp.ClientTimeout(total=timeout),
        ) as client:
            try:
                resp = await client.get(fsapi_device_url)
                doc = ET.fromstring(await resp.text(encoding="utf-8"))

                api = doc.find("webfsapi")
                if api is not None and api.text:
                    return api.text
                else:
                    raise FSApiException(
                        f"Could not retrieve webfsapi endpoint from {fsapi_device_url}"
                    )

            except (aiohttp.ServerTimeoutError, asyncio.TimeoutError):
                raise ConnectionError(
                    f"Did not get a response in time from {fsapi_device_url}"
                )
            except aiohttp.ClientConnectionError:
                raise ConnectionError(f"Could not connect to {fsapi_device_url}")

    @staticmethod
    async def create(
        fsapi_device_url: str,
        pin: t.Union[str, int],
        timeout: int = DEFAULT_TIMEOUT_IN_SECONDS,
    ) -> "AFSAPI":
        webfsapi_endpoint = await AFSAPI.get_webfsapi_endpoint(
            fsapi_device_url, timeout
        )

        return AFSAPI(webfsapi_endpoint, pin, timeout)

    # http request helpers
    async def _create_session(self) -> t.Optional[str]:
        self.sid = None
        return unpack_xml(
            await self.__call("CREATE_SESSION", retry_with_session=False), "sessionId"
        )

    async def __call(
        self,
        path: str,
        extra: t.Optional[t.Dict[str, DataItem]] = None,
        force_new_session: bool = False,
        retry_with_session: bool = True,
        throttle_wait_after_call: float = TIME_AFTER_READ_CALLS_IN_SECONDS,
    ) -> ET.Element:
        """Execute a frontier silicon API call."""

        params: t.Dict[str, DataItem] = dict(pin=self.pin)

        if force_new_session:
            self.sid = await self._create_session()
        if self.sid:
            params.update(sid=self.sid)

        if extra:
            params.update(**extra)

        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(force_close=True),
            timeout=aiohttp.ClientTimeout(total=self.timeout),
        ) as client:
            try:
                async with self.__throttler.throttle(throttle_wait_after_call):
                    result = await client.get(
                        f"{self.webfsapi_endpoint}/{path}", params=params
                    )

                LOGGER.debug(f"Called {path} with {params}: {result.status}")

                if result.status == 403:
                    raise InvalidPinException("Access denied - incorrect PIN")
                elif result.status == 404:
                    # Bad session ID or service endpoint
                    logging.warn(
                        f"Service call failed with 404 to {self.webfsapi_endpoint}/{path}"
                    )

                    if not force_new_session and retry_with_session:
                        # retry command with a forced new session
                        return await self.__call(path, extra, force_new_session=True)
                    else:
                        raise InvalidSessionException(
                            "Wrong session-id or invalid command"
                        )
                elif result.status != 200:
                    raise FSApiException(
                        f"Unexpected result {result.status}: {await result.text()}"
                    )
                doc = ET.fromstring(await result.text(encoding="utf-8"))
                status = unpack_xml(doc, "status")

                if status == "FS_OK" or status == "FS_LIST_END":
                    return doc
                elif status == "FS_NODE_DOES_NOT_EXIST":
                    raise NotImplementedException(
                        f"FSAPI service {path} not implemented at {self.webfsapi_endpoint}."
                    )
                elif status == "FS_NODE_BLOCKED":
                    raise FSApiException("Device is not in the correct mode")
                elif status == "FS_FAIL":
                    raise OutOfRangeException(
                        "Command failed. Value is not in range for this command."
                    )
                elif status == "FS_PACKET_BAD":
                    raise FSApiException("This command can't be SET")

                logging.error(f"Unexpected FSAPI status {status}")
                raise FSApiException(f"Unexpected FSAPI status '{status}'")
            except aiohttp.ClientConnectionError:
                raise ConnectionError(f"Could not connect to {self.webfsapi_endpoint}")
            except TimeoutError:
                if not force_new_session and retry_with_session:
                    return await self.__call(path, extra, force_new_session=True)
                else:
                    raise ConnectionError(
                        f"{self.webfsapi_endpoint} did not respond within {self.timeout} seconds"
                    )

    # Helper methods

    # Handlers

    async def handle_get(self, item: str) -> ET.Element:
        return await self.__call(f"GET/{item}")

    async def handle_set(
        self,
        item: str,
        value: t.Any,
        throttle_wait_after_call: float = TIME_AFTER_SET_CALLS_IN_SECONDS,
    ) -> t.Optional[bool]:
        status = unpack_xml(
            await self.__call(
                f"SET/{item}",
                dict(value=value),
                throttle_wait_after_call=throttle_wait_after_call,
            ),
            "status",
        )
        return maybe(status, lambda x: x == "FS_OK")

    async def handle_text(self, item: str) -> t.Optional[str]:
        return unpack_xml(await self.handle_get(item), "value/c8_array")

    async def handle_int(self, item: str) -> t.Optional[int]:
        val = unpack_xml(await self.handle_get(item), "value/u8")
        return maybe(val, int)

    # returns an int, assuming the value does not exceed 8 bits
    async def handle_long(self, item: str) -> t.Optional[int]:
        val = unpack_xml(await self.handle_get(item), "value/u32")
        return maybe(val, int)

    async def handle_signed_long(
        self,
        item: str,
    ) -> t.Optional[int]:
        val = unpack_xml(await self.handle_get(item), "value/s32")
        return maybe(val, int)

    async def handle_list(
        self, list_name: str
    ) -> t.AsyncIterable[t.Tuple[str, t.Dict[str, t.Optional[DataItem]]]]:
        def _handle_item(
            item: ET.Element,
        ) -> t.Tuple[str, t.Dict[str, t.Optional[DataItem]]]:
            key = item.attrib["key"]

            def _handle_field(field: ET.Element) -> t.Tuple[str, t.Optional[DataItem]]:
                # TODO: Handle other field types
                if "name" in field.attrib:
                    id = field.attrib["name"]
                    s = unpack_xml(field, "c8_array")
                    v = maybe(unpack_xml(field, "u8"), int)
                    return (id, s or v)
                raise ValueError("Invalid field")

            value = dict(map(_handle_field, item.findall("field")))
            return key, value

        async def _get_next_items(
            start: int, count: int
        ) -> t.Tuple[list[ET.Element], bool]:
            try:
                doc = await self.__call(
                    f"LIST_GET_NEXT/{list_name}/{start}", {"maxItems": count}
                )

                if doc and unpack_xml(doc, "status") == "FS_OK":
                    return doc.findall("item"), doc.find("listend") is not None
                else:
                    return [], True
            except OutOfRangeException:
                return [], True

        start = -1
        count = 50  # asking for more items gives a bigger chance on FS_NODE_BLOCKED errors on subsequent requests
        has_next = True

        while has_next:
            items, end_reached = await _get_next_items(start, count)

            for item in items:
                yield _handle_item(item)

            start += count

            if end_reached:
                has_next = False

    # sys
    async def get_friendly_name(self) -> t.Optional[str]:
        """Get the friendly name of the device."""
        return await self.handle_text(API["friendly_name"])

    async def set_friendly_name(self, value: str) -> t.Optional[bool]:
        """Set the friendly name of the device."""
        return await self.handle_set(API["friendly_name"], value)

    async def get_version(self) -> t.Optional[str]:
        """Get the friendly name of the device."""
        return await self.handle_text(API["version"])

    async def get_radio_id(self) -> t.Optional[str]:
        """Get the friendly name of the device."""
        return await self.handle_text(API["radio_id"])

    async def get_mac(self) -> t.Optional[str]:
        """Get the MAC address of the device."""
        on_wlan = await self.handle_int(API["wlan_active"])
        if bool(on_wlan):
            return await self.handle_text(API["wlan_mac"])
        else:
            return await self.handle_text(API["wired_mac"])

    async def get_rssi(self) -> t.Optional[int]:
        """Get the current wlan Received Signal Strength Indication in dBm"""

        # RSSI is returned as a percentage by the API, scaled linearly between
        # -80dBm (0%) and -20dBm (100%).  100% indicates a wired
        # connection.  This functions returns the dBm value of RSSI.

        rssi = await self.handle_int(API["rssi"])
        if rssi is not None:
            return int(round(rssi * 0.6 - 80))
        else:
            return None

    async def get_power(self) -> t.Optional[bool]:
        """Check if the device is on."""
        power = await self.handle_int(API["power"])
        return bool(power)

    async def set_power(self, value: bool = False) -> t.Optional[bool]:
        """Power on or off the device."""
        power = await self.handle_set(
            API["power"],
            int(value),
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )
        return bool(power)

    async def get_volume_steps(self) -> t.Optional[int]:
        """Read the maximum volume level of the device."""
        if not self.__volume_steps:
            self.__volume_steps = await self.handle_int(API["volume_steps"])

        return self.__volume_steps

    # Volume
    async def get_volume(self) -> t.Optional[int]:
        """Read the volume level of the device."""
        return await self.handle_int(API["volume"])

    async def set_volume(self, value: int) -> t.Optional[bool]:
        """Set the volume level of the device."""
        return await self.handle_set(API["volume"], value)

    # Mute
    async def get_mute(self) -> t.Optional[bool]:
        """Check if the device is muted."""
        mute = await self.handle_int(API["mute"])
        return bool(mute)

    async def set_mute(self, value: bool = False) -> t.Optional[bool]:
        """Mute or unmute the device."""
        mute = await self.handle_set(API["mute"], int(value))
        return bool(mute)

    async def get_play_status(self) -> t.Optional[PlayState]:
        """Get the play status of the device."""
        status = await self.handle_int(API["status"])
        if status:
            return PlayState(status)
        else:
            return None

    async def get_play_name(self) -> t.Optional[str]:
        """Get the name of the played item."""
        return await self.handle_text(API["name"])

    async def get_play_text(self) -> t.Optional[str]:
        """Get the text associated with the played media."""
        return await self.handle_text(API["text"])

    async def get_play_artist(self) -> t.Optional[str]:
        """Get the artists of the current media(song)."""
        return await self.handle_text(API["artist"])

    async def get_play_album(self) -> t.Optional[str]:
        """Get the songs's album."""
        return await self.handle_text(API["album"])

    async def get_play_graphic(self) -> t.Optional[str]:
        """Get the album art associated with the song/album/artist."""
        return await self.handle_text(API["graphic_uri"])

    # Shuffle
    async def get_play_shuffle(self) -> t.Optional[bool]:
        status = await self.handle_int(API["shuffle"])
        if status:
            return status == 1
        return None

    async def set_play_shuffle(self, value: bool) -> t.Optional[bool]:
        return await self.handle_set(API["shuffle"], int(value))

    # Repeat
    async def get_play_repeat(self) -> t.Optional[bool]:
        status = await self.handle_int(API["repeat"])
        if status:
            return status == 1
        return None

    async def play_repeat(self, value: bool) -> t.Optional[bool]:
        return await self.handle_set(API["repeat"], int(value))

    async def get_play_duration(self) -> t.Optional[int]:
        """Get the duration of the played media."""
        return await self.handle_long(API["duration"])

    async def get_play_position(self) -> t.Optional[int]:
        """
        The user can jump to a specific moment of the track. This means that the range of the value is
        different with every track.
        After the position is changed, the music player will continue to play the song (with the same rate).

        To find the upper bound for the current track, use `get_play_duration`
        """
        return await self.handle_int(API["position"])

    async def set_play_position(self, value: int) -> t.Optional[bool]:
        return await self.handle_set(API["position"], value)

    # Play  rate
    async def get_play_rate(self) -> t.Optional[int]:
        """
        * -127 to -1: When the user sends a negative value, the music player will rewind the track.
          The speed depends on the value, -10 will rewind faster than value -2

        * 0: When the user send rate = 0, the track will be paused
        * 1: The track will be played with normal speed
        * 2 to 127: The track will be fast forwarded, the speed is here also dependable of the value
          The speed of the fast forward is also dependable of the value, 80 is faster than 10
        """
        return await self.handle_int(API["rate"])

    async def set_play_rate(self, value: int) -> t.Optional[bool]:
        if -127 <= value <= 127:
            return await self.handle_set(API["rate"], value)
        else:
            raise ValueError("Play rate must be within values -127 to 127")

    # play controls

    async def play_control(self, value: t.Union[PlayControl, int]) -> t.Optional[bool]:
        """
        Control the player of the device.

        1=Play; 2=Pause; 3=Next; 4=Previous (song/station)
        """
        return await self.handle_set(API["control"], int(value))

    async def play(self) -> t.Optional[bool]:
        """Play media."""
        mode = await self.get_mode()
        if mode.label == "Internet radio":
            return await self.play_control(PlayControl.STOP)
        return await self.play_control(PlayControl.PLAY)

    async def pause(self) -> t.Optional[bool]:
        """Pause playing."""
        mode = await self.get_mode()
        if mode.label == "Internet radio":
            return await self.play_control(PlayControl.STOP)
        return await self.play_control(PlayControl.PAUSE)

    async def forward(self) -> t.Optional[bool]:
        """Next media."""
        mode = await self.get_mode()
        if mode.label == "Internet radio":
            return None #Not supported
        return await self.play_control(PlayControl.NEXT)

    async def rewind(self) -> t.Optional[bool]:
        """Previous media."""
        mode = await self.get_mode()
        if mode.label == "Internet radio":
            return None #Not supported
        return await self.play_control(PlayControl.PREV)

    async def get_equalisers(self) -> t.List[Equaliser]:
        """Get the equaliser modes supported by this device."""

        # Cache as this never changes
        if self.__equalisers is None:
            self.__equalisers = [
                Equaliser(key=key, **eqinfo)  # type: ignore
                async for key, eqinfo in self.handle_list(API["equalisers"])
            ]

        return self.__equalisers  # type: ignore

    # EQ Presets
    async def get_eq_preset(self) -> t.Optional[Equaliser]:
        v = await self.handle_int(API["eqpreset"])
        if v is None:
            return None

        for eq in await self.get_equalisers():
            if eq.key == str(v):
                return eq

        raise FSApiException(f"Could not retrieve equaliser {v} in equaliser list")

    async def set_eq_preset(self, value: t.Union[Equaliser, int]) -> t.Optional[bool]:
        return await self.handle_set(
            API["eqpreset"],
            int(value.key) if isinstance(value, Equaliser) else value,
        )

    # EQ Loudness (Only works with My EQ!)
    async def get_eq_loudness(self) -> bool:
        return bool(await self.handle_int(API["eqloudness"]))

    async def set_eq_loudness(self, value: bool) -> t.Optional[bool]:
        return await self.handle_set(API["eqloudness"], int(value))

    # Bass and Treble
    async def get_bass(self) -> t.Optional[int]:
        return await self.handle_int(API["bass"])

    async def set_bass(self, value: bool) -> t.Optional[bool]:
        if -14 <= value <= 14:
            return await self.handle_set(API["bass"], int(value))
        else:
            raise ValueError("Outside of bounds: [-14, 14]")

    async def get_treble(self) -> t.Optional[int]:
        return await self.handle_int(API["treble"])

    async def set_treble(self, value: bool) -> t.Optional[bool]:
        if -14 <= value <= 14:
            return await self.handle_set(API["treble"], int(value))
        else:
            raise ValueError("Outside of bounds: [-14, 14]")

    # Mode
    async def _get_modes(
        self,
    ) -> t.AsyncIterable[t.Tuple[str, t.Dict[str, t.Optional[DataItem]]]]:
        async for mode in self.handle_list(API["valid_modes"]):
            yield mode

    async def get_modes(self) -> t.List[PlayerMode]:
        """Get the modes supported by this device."""

        # Cache as this never changes
        if self.__modes is None:
            self.__modes = [
                PlayerMode(key=k, **v)  # type: ignore
                async for k, v in self._get_modes()
            ]

        return self.__modes  # type: ignore

    async def get_mode(self) -> t.Optional[PlayerMode]:
        """Get the currently active mode on the device (DAB, FM, Spotify)."""
        int_mode = await self.handle_long(API["mode"])
        if int_mode is None:
            return None

        for mode in await self.get_modes():
            if mode.key == str(int_mode):
                return mode

        raise FSApiException(f"Could not retrieve mode {int_mode} in modes list")

    async def set_mode(self, value: t.Union[PlayerMode, str]) -> t.Optional[bool]:
        """Set the currently active mode on the device (DAB, FM, Spotify)."""
        result = await self.handle_set(
            API["mode"],
            value.key if isinstance(value, PlayerMode) else value,
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )
        self._current_nav_path = []
        return result

    # Sleep
    async def get_sleep(self) -> t.Optional[int]:
        """Check when and if the device is going to sleep."""
        return await self.handle_long(API["sleep"])

    async def set_sleep(self, value: int = 0) -> t.Optional[bool]:
        """Set device sleep timer."""
        return await self.handle_set(API["sleep"], int(value))

    # Folder navigation

    async def _enable_nav_if_necessary(self) -> None:
        nav_state = await self.handle_int(API["nav_state"])
        if nav_state != 1:
            await self.handle_set(
                API["nav_state"],
                1,
                throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
                # changing to navigation can be very slow!
            )

            # the nav path is empty, as we needed to set the radio into nav-mode
            self._current_nav_path = []

    async def nav_get_numitems(self) -> t.Optional[int]:
        await self._enable_nav_if_necessary()
        return await self.handle_signed_long(API["numitems"])

    async def nav_list(
        self,
    ) -> t.AsyncIterable[t.Tuple[str, t.Dict[str, t.Optional[DataItem]]]]:
        await self._enable_nav_if_necessary()
        return self.handle_list(API["nav_list"])

    async def nav_select_folder(self, value: int) -> t.Optional[bool]:
        await self._enable_nav_if_necessary()
        result = await self.handle_set(
            API["navigate"],
            value,
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )
        self._current_nav_path.append(value)

        return result

    async def nav_select_parent_folder(self) -> t.Optional[bool]:
        await self._enable_nav_if_necessary()
        result = await self.handle_set(
            API["navigate"],
            "0xffffffff",
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )
        self._current_nav_path.pop()

        return result

    async def nav_select_item(self, value: int) -> t.Optional[bool]:
        await self._enable_nav_if_necessary()
        return await self.handle_set(
            API["selectItem"],
            value,
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )

    async def nav_reset(self) -> t.Optional[bool]:
        self._current_nav_path = []
        return await self.handle_set(API["nav_state"], 0)

    async def nav_select_folder_via_path(self, path: list[int]) -> t.Optional[bool]:
        """Navigates to a target folder from the current folder in as litte steps as necessary."""
        result = None

        LOGGER.debug("Navigating to %s, currently in %s", path, self._current_nav_path)

        while len(self._current_nav_path) > len(path):
            LOGGER.debug("Going up to parent folder in %s", self._current_nav_path)
            result = await self.nav_select_parent_folder()

        key_idx = 0
        while key_idx < len(path):
            key = int(path[key_idx])
            if key_idx >= len(self._current_nav_path):
                LOGGER.debug("Selecting %s in %s", key, self._current_nav_path)
                result = await self.nav_select_folder(key)
                key_idx += 1
            elif key != self._current_nav_path[key_idx]:
                LOGGER.debug("Going up to parent folder in %s", self._current_nav_path)
                result = await self.nav_select_parent_folder()
            else:
                key_idx += 1

        return result

    async def nav_select_item_via_path(self, path: list[int]) -> t.Optional[bool]:
        await self.nav_select_folder_via_path(path[:-1])
        return await self.nav_select_item(path[-1])

    # Presets

    async def _get_presets(
        self,
    ) -> t.AsyncIterable[t.Tuple[str, t.Dict[str, t.Optional[DataItem]]]]:
        await self._enable_nav_if_necessary()

        async for key, preset in self.handle_list(API["presets"]):
            if preset.get("name"):
                # Strip whitespaces from names
                assert isinstance(preset["name"], str)
                preset["name"] = preset["name"].strip()
                yield key, preset
            else:
                # Skip empty preset
                pass

    async def get_presets(self) -> t.List[Preset]:

        # We don't cache this call as it changes when the mode changes

        def _to_preset(
            key: str, preset_fields: t.Dict[str, t.Optional[DataItem]]
        ) -> Preset:
            assert isinstance(preset_fields["name"], str)
            type = str(preset_fields["type"]) if "type" in preset_fields else None
            return Preset(int(key), type, preset_fields["name"])

        return [
            _to_preset(key, preset_fields)
            async for key, preset_fields in self._get_presets()
        ]

    async def select_preset(self, value: t.Union[Preset, int]) -> t.Optional[bool]:
        await self._enable_nav_if_necessary()
        return await self.handle_set(
            API["selectPreset"],
            value.key if isinstance(value, Preset) else value,
            throttle_wait_after_call=TIME_AFTER_SLOW_SET_CALLS_IN_SECONDS,
        )