Skip to content
interfaces.py 2.32 KiB
Newer Older
#  Copyright 2019  Dom Sekotill <dom.sekotill@kodo.org.uk>
#
#  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 Dict
from typing import Iterable
from .. import config
from .. import util
from ..types import PathLike
from ..types import StringMap
from . import consts
from .base import BaseClient


class InterfaceClient(BaseClient):
	"""
	A client for per-interface management
	"""
	async def connect(self, path: PathLike):
		await super().connect(path)
		self.name = await self.send_command(consts.COMMAND_IFNAME, convert=str)
	async def scan(self) -> Iterable[StringMap]:
		"""
		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():
Dom Sekotill's avatar
Dom Sekotill committed
				bss = await self.send_command(
					consts.COMMAND_BSS, str(idx), convert=util.kv2dict,
Dom Sekotill's avatar
Dom Sekotill committed
				)
				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):
		"""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,
		)