# Copyright 2019-2021 Dom Sekotill # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Interfaces control client class """ from itertools import count from typing import Any from typing import AsyncGenerator from typing import Dict from .. import config from ..types import PathLike from . import consts from .base import BaseClient StringMap = Dict[str, str] class InterfaceClient(BaseClient): """ A client for per-interface management """ name = None async def connect(self, path: PathLike) -> None: await super().connect(path) self.name = await self.send_command(consts.COMMAND_IFNAME, convert=str) async def scan(self) -> AsyncGenerator[StringMap, None]: """ Iteratively produces the details of all detectable IEEE 802.11 BSS (WiFi Access Points to you and me) """ async with self.attach(): await self.send_command(consts.COMMAND_SCAN) await self.event(consts.CTRL_EVENT_SCAN_RESULTS) for idx in count(): bss = await self.send_command( consts.COMMAND_BSS, str(idx), convert=_kv2dict, ) if not bss: return yield bss async def add_network(self, configuration: Dict[str, Any]) -> int: """Add a new network configuration""" netid = await self.send_command(consts.COMMAND_ADD_NETWORK, convert=str) for var, val in configuration.items(): await self.set_network(netid, var, val) await self.send_command(consts.COMMAND_ENABLE_NETWORK, netid) return int(netid) async def set_network(self, netid: str, variable: str, value: Any) -> None: """Set a network configuration option""" if not isinstance(value, config.get_type(variable)): raise TypeError(f"Wrong type for {variable}: {value!r}") await self.send_command( consts.COMMAND_SET_NETWORK, netid, variable, f'"{value}"' if isinstance(value, str) else str(value), separator=consts.SEPARATOR_SPACE, ) def _kv2dict(keyvalues: str) -> StringMap: """ Convert a list of line-terminated "key=value" substrings into a dictionary """ return dict(kv.split("=", 1) for kv in keyvalues.splitlines()) # type: ignore