# Copyright 2019-2021, 2024 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 __future__ import annotations from itertools import count from os import PathLike from typing import AsyncGenerator from typing import Dict from .. import config 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[str]) -> 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, object]) -> 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: object) -> 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())