zigpy / bellows / 989

pending completion

Build # 989

Build Type

Pull #280

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/zigpy/bellows/commit/<a class=hub.com/zigpy/bellows/commit/20674bca4f8918478f40e11714707084e36ba0f2">20674bca4<a href="https://github.com/zigpy/bellows/commit/20674bca4f8918478f40e11714707084e36ba0f2">">Merge </a><a class="double-link" href="https://github.com/zigpy/bellows/commit/<a class="double-link" href="https://github.com/zigpy/bellows/commit/9a1ba35c1ccc272f61e488625ffe58d2019c0d1b">9a1ba35c1</a>">9a1ba35c1</a><a href="https://github.com/zigpy/bellows/commit/20674bca4f8918478f40e11714707084e36ba0f2"> into 5874e59e1">5874e59e1</a>

Pull Request Pull Request #280: Refactor EZSP to support multiple protocol versions

Run Details

440 of 440 new or added lines in 13 files covered. (100.0%)

2027 of 2067 relevant lines covered (98.06%)

1.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.63

/bellows/ezsp/__init__.py

"""EZSP protocol."""

import asyncio
import functools
import logging
from typing import Any, Callable, Coroutine, Dict, Tuple

from bellows.config import (
    CONF_DEVICE,
    CONF_DEVICE_PATH,
    CONF_EZSP_CONFIG,
    SCHEMA_DEVICE,
)
from bellows.exception import APIException, EzspError
import bellows.ezsp.v4
import bellows.types as t
import bellows.uart
import serial

PROBE_TIMEOUT = 3
LOGGER = logging.getLogger(__name__)

PROTOCOL_BY_VER = {}


class EZSP:
    def __init__(self, device_config: Dict):
        self._config = device_config
        self._callbacks = {}
        self._ezsp_event = asyncio.Event()
        self._ezsp_version = bellows.ezsp.v4.EZSP_VERSION
        self._gw = None
        self._protocol = None

    @classmethod
    async def probe(cls, device_config: Dict) -> bool:
        """Probe port for the device presence."""
        ezsp = cls(SCHEMA_DEVICE(device_config))
        try:
            await asyncio.wait_for(ezsp._probe(), timeout=PROBE_TIMEOUT)
            return True
        except (asyncio.TimeoutError, serial.SerialException, APIException) as exc:
            LOGGER.debug(
                "Unsuccessful radio probe of '%s' port",
                device_config[CONF_DEVICE_PATH],
                exc_info=exc,
            )
        finally:
            ezsp.close()

        return False

    async def _probe(self) -> None:
        """Open port and try sending a command"""
        await self.connect()
        await self.reset()
        self.close()

    @classmethod
    async def initialize(cls, zigpy_config: Dict) -> "EZSP":
        """Return initialized EZSP instance. """
        ezsp = cls(zigpy_config[CONF_DEVICE])
        await ezsp.connect()
        await ezsp.reset()
        await ezsp.version()
        await ezsp._protocol.initialize(zigpy_config[CONF_EZSP_CONFIG])
        return ezsp

    async def connect(self) -> None:
        assert self._gw is None
        self._gw = await bellows.uart.connect(self._config, self)
        self._protocol = bellows.ezsp.v4.EZSPv4(self.handle_callback, self._gw)

    async def reset(self):
        LOGGER.debug("Resetting EZSP")
        self.stop_ezsp()
        await self._gw.reset()
        self.start_ezsp()

    async def version(self):
        ver, stack_type, stack_version = await self._command(
            "version", self.ezsp_version
        )
        if ver != self.ezsp_version:
            self._ezsp_version = ver
            LOGGER.debug("Switching to EZSP protocol version %d", self.ezsp_version)
            try:
                protcol_cls = PROTOCOL_BY_VER[ver]
            except KeyError:
                LOGGER.warning(
                    "Protocol version %s is not supported, using version %s instead",
                    ver,
                    bellows.ezsp.v4.EZSP_VERSION,
                )
                protcol_cls = bellows.ezsp.v4.EZSPv4
            self._protocol = protcol_cls(self.handle_callback, self._gw)
            await self._command("version", ver)
        LOGGER.debug(
            "EZSP Stack Type: %s, Stack Version: %s, Protocol version: %s",
            stack_type,
            stack_version,
            ver,
        )

    def close(self):
        self.stop_ezsp()
        if self._gw:
            self._gw.close()
            self._gw = None

    def _command(self, name: str, *args: Tuple[Any, ...]) -> asyncio.Future:
        if not self.is_ezsp_running:
            LOGGER.debug(
                "Couldn't send command %s(%s). EZSP is not running", name, args
            )
            raise EzspError("EZSP is not running")

        return self._protocol.command(name, *args)

    async def _list_command(self, name, item_frames, completion_frame, spos, *args):
        """Run a command, returning result callbacks as a list"""
        fut = asyncio.Future()
        results = []

        def cb(frame_name, response):
            if frame_name in item_frames:
                results.append(response)
            elif frame_name == completion_frame:
                fut.set_result(response)

        cbid = self.add_callback(cb)
        try:
            v = await self._command(name, *args)
            if v[0] != t.EmberStatus.SUCCESS:
                raise Exception(v)
            v = await fut
            if v[spos] != t.EmberStatus.SUCCESS:
                raise Exception(v)
        finally:
            self.remove_callback(cbid)

        return results

    startScan = functools.partialmethod(
        _list_command,
        "startScan",
        ["energyScanResultHandler", "networkFoundHandler"],
        "scanCompleteHandler",
        1,
    )
    pollForData = functools.partialmethod(
        _list_command, "pollForData", ["pollHandler"], "pollCompleteHandler", 0
    )
    zllStartScan = functools.partialmethod(
        _list_command,
        "zllStartScan",
        ["zllNetworkFoundHandler"],
        "zllScanCompleteHandler",
        0,
    )
    rf4ceDiscovery = functools.partialmethod(
        _list_command,
        "rf4ceDiscovery",
        ["rf4ceDiscoveryResponseHandler"],
        "rf4ceDiscoveryCompleteHandler",
        0,
    )

    def connection_lost(self, exc):
        """Lost serial connection."""
        LOGGER.debug(
            "%s connection lost unexpectedly: %s", self._config[CONF_DEVICE_PATH], exc
        )
        self.enter_failed_state("Serial connection loss: {}".format(exc))

    def enter_failed_state(self, error):
        """UART received error frame."""
        LOGGER.error("NCP entered failed state. Requesting APP controller restart")
        self.close()
        self.handle_callback("_reset_controller_application", (error,))

    async def formNetwork(self, parameters):  # noqa: N802
        fut = asyncio.Future()

        def cb(frame_name, response):
            nonlocal fut
            if frame_name == "stackStatusHandler":
                fut.set_result(response)

        self.add_callback(cb)
        v = await self._command("formNetwork", parameters)
        if v[0] != t.EmberStatus.SUCCESS:
            raise Exception("Failure forming network: %s" % (v,))

        v = await fut
        if v[0] != t.EmberStatus.NETWORK_UP:
            raise Exception("Failure forming network: %s" % (v,))

        return v

    def __getattr__(self, name: str) -> Callable:
        if name not in self._protocol.COMMANDS:
            raise AttributeError(f"{name} not found in COMMANDS")

        return functools.partial(self._command, name)

    def frame_received(self, data: bytes) -> None:
        """Handle a received EZSP frame

        The protocol has taken care of UART specific framing etc, so we should
        just have EZSP application stuff here, with all escaping/stuffing and
        data randomization removed.
        """
        self._protocol(data)

    def add_callback(self, cb):
        id_ = hash(cb)
        while id_ in self._callbacks:
            id_ += 1
        self._callbacks[id_] = cb
        return id_

    def remove_callback(self, id_):
        return self._callbacks.pop(id_)

    def handle_callback(self, *args):
        for callback_id, handler in self._callbacks.items():
            try:
                handler(*args)
            except Exception as e:
                LOGGER.exception("Exception running handler", exc_info=e)

    def update_policies(self, zigpy_config: dict) -> Coroutine:
        """Set up the policies for what the NCP should do."""
        return self._protocol.update_policies(zigpy_config)

    def start_ezsp(self):
        """Mark EZSP as running."""
        self._ezsp_event.set()

    def stop_ezsp(self):
        """Mark EZSP stopped."""
        self._ezsp_event.clear()

    @property
    def is_ezsp_running(self):
        """Return True if EZSP is running."""
        return self._ezsp_event.is_set()

    @property
    def ezsp_version(self):
        """Return protocol version."""
        return self._ezsp_version

    @property
    def types(self):
        """Return EZSP types for this specific version."""
        return self._protocol.types

1	"""EZSP protocol."""
2
3	import asyncio	2✔
4	import functools	2✔
5	import logging	2✔
6	from typing import Any, Callable, Coroutine, Dict, Tuple	2✔
7
8	from bellows.config import (	2✔
9	CONF_DEVICE,
10	CONF_DEVICE_PATH,
11	CONF_EZSP_CONFIG,
12	SCHEMA_DEVICE,
13	)
14	from bellows.exception import APIException, EzspError	2✔
15	import bellows.ezsp.v4	2✔
16	import bellows.types as t	2✔
17	import bellows.uart	2✔
18	import serial	2✔
19
20	PROBE_TIMEOUT = 3	2✔
21	LOGGER = logging.getLogger(__name__)	2✔
22
23	PROTOCOL_BY_VER = {}	2✔
24
25
26	class EZSP:	2✔
27	def __init__(self, device_config: Dict):	2✔
28	self._config = device_config	2✔
29	self._callbacks = {}	2✔
30	self._ezsp_event = asyncio.Event()	2✔
31	self._ezsp_version = bellows.ezsp.v4.EZSP_VERSION	2✔
32	self._gw = None	2✔
33	self._protocol = None	2✔
34
35	@classmethod	2✔
36	async def probe(cls, device_config: Dict) -> bool:	2✔
37	"""Probe port for the device presence."""
38	ezsp = cls(SCHEMA_DEVICE(device_config))	2✔
39	try:	2✔
40	await asyncio.wait_for(ezsp._probe(), timeout=PROBE_TIMEOUT)	2✔
41	return True	2✔
42	except (asyncio.TimeoutError, serial.SerialException, APIException) as exc:	2✔
43	LOGGER.debug(	2✔
44	"Unsuccessful radio probe of '%s' port",
45	device_config[CONF_DEVICE_PATH],
46	exc_info=exc,
47	)
48	finally:
49	ezsp.close()	2✔
50
51	return False	2✔
52
53	async def _probe(self) -> None:	2✔
54	"""Open port and try sending a command"""
55	await self.connect()	2✔
56	await self.reset()	2✔
57	self.close()	2✔
58
59	@classmethod	2✔
60	async def initialize(cls, zigpy_config: Dict) -> "EZSP":	2✔
61	"""Return initialized EZSP instance. """
62	ezsp = cls(zigpy_config[CONF_DEVICE])	×
63	await ezsp.connect()	×
64	await ezsp.reset()	×
65	await ezsp.version()	×
66	await ezsp._protocol.initialize(zigpy_config[CONF_EZSP_CONFIG])	×
67	return ezsp	×
68
69	async def connect(self) -> None:	2✔
70	assert self._gw is None	2✔
71	self._gw = await bellows.uart.connect(self._config, self)	2✔
72	self._protocol = bellows.ezsp.v4.EZSPv4(self.handle_callback, self._gw)	2✔
73
74	async def reset(self):	2✔
75	LOGGER.debug("Resetting EZSP")	2✔
76	self.stop_ezsp()	2✔
77	await self._gw.reset()	2✔
78	self.start_ezsp()	2✔
79
80	async def version(self):	2✔
81	ver, stack_type, stack_version = await self._command(	2✔
82	"version", self.ezsp_version
83	)
84	if ver != self.ezsp_version:	2✔
85	self._ezsp_version = ver	2✔
86	LOGGER.debug("Switching to EZSP protocol version %d", self.ezsp_version)	2✔
87	try:	2✔
88	protcol_cls = PROTOCOL_BY_VER[ver]	2✔
89	except KeyError:	2✔
90	LOGGER.warning(	2✔
91	"Protocol version %s is not supported, using version %s instead",
92	ver,
93	bellows.ezsp.v4.EZSP_VERSION,
94	)
95	protcol_cls = bellows.ezsp.v4.EZSPv4	2✔
96	self._protocol = protcol_cls(self.handle_callback, self._gw)	2✔
97	await self._command("version", ver)	2✔
98	LOGGER.debug(	2✔
99	"EZSP Stack Type: %s, Stack Version: %s, Protocol version: %s",
100	stack_type,
101	stack_version,
102	ver,
103	)
104
105	def close(self):	2✔
106	self.stop_ezsp()	2✔
107	if self._gw:	2✔
108	self._gw.close()	2✔
109	self._gw = None	2✔
110
111	def _command(self, name: str, *args: Tuple[Any, ...]) -> asyncio.Future:	2✔
112	if not self.is_ezsp_running:	2✔
113	LOGGER.debug(	2✔
114	"Couldn't send command %s(%s). EZSP is not running", name, args
115	)
116	raise EzspError("EZSP is not running")	2✔
117
118	return self._protocol.command(name, *args)	2✔
119
120	async def _list_command(self, name, item_frames, completion_frame, spos, *args):	2✔
121	"""Run a command, returning result callbacks as a list"""
122	fut = asyncio.Future()	2✔
123	results = []	2✔
124
125	def cb(frame_name, response):	2✔
126	if frame_name in item_frames:	2✔
127	results.append(response)	2✔
128	elif frame_name == completion_frame:	2✔
129	fut.set_result(response)	2✔
130
131	cbid = self.add_callback(cb)	2✔
132	try:	2✔
133	v = await self._command(name, *args)	2✔
134	if v[0] != t.EmberStatus.SUCCESS:	2✔
135	raise Exception(v)	2✔
136	v = await fut	2✔
137	if v[spos] != t.EmberStatus.SUCCESS:	2✔
138	raise Exception(v)	2✔
139	finally:
140	self.remove_callback(cbid)	2✔
141
142	return results	2✔
143
144	startScan = functools.partialmethod(	2✔
145	_list_command,
146	"startScan",
147	["energyScanResultHandler", "networkFoundHandler"],
148	"scanCompleteHandler",
149	1,
150	)
151	pollForData = functools.partialmethod(	2✔
152	_list_command, "pollForData", ["pollHandler"], "pollCompleteHandler", 0
153	)
154	zllStartScan = functools.partialmethod(	2✔
155	_list_command,
156	"zllStartScan",
157	["zllNetworkFoundHandler"],
158	"zllScanCompleteHandler",
159	0,
160	)
161	rf4ceDiscovery = functools.partialmethod(	2✔
162	_list_command,
163	"rf4ceDiscovery",
164	["rf4ceDiscoveryResponseHandler"],
165	"rf4ceDiscoveryCompleteHandler",
166	0,
167	)
168
169	def connection_lost(self, exc):	2✔
170	"""Lost serial connection."""
171	LOGGER.debug(	2✔
172	"%s connection lost unexpectedly: %s", self._config[CONF_DEVICE_PATH], exc
173	)
174	self.enter_failed_state("Serial connection loss: {}".format(exc))	2✔
175
176	def enter_failed_state(self, error):	2✔
177	"""UART received error frame."""
178	LOGGER.error("NCP entered failed state. Requesting APP controller restart")	2✔
179	self.close()	2✔
180	self.handle_callback("_reset_controller_application", (error,))	2✔
181
182	async def formNetwork(self, parameters): # noqa: N802	2✔
183	fut = asyncio.Future()	2✔
184
185	def cb(frame_name, response):	2✔
186	nonlocal fut
187	if frame_name == "stackStatusHandler":	2✔
188	fut.set_result(response)	2✔
189
190	self.add_callback(cb)	2✔
191	v = await self._command("formNetwork", parameters)	2✔
192	if v[0] != t.EmberStatus.SUCCESS:	2✔
193	raise Exception("Failure forming network: %s" % (v,))	2✔
194
195	v = await fut	2✔
196	if v[0] != t.EmberStatus.NETWORK_UP:	2✔
197	raise Exception("Failure forming network: %s" % (v,))	2✔
198
199	return v	2✔
200
201	def __getattr__(self, name: str) -> Callable:	2✔
202	if name not in self._protocol.COMMANDS:	2✔
203	raise AttributeError(f"{name} not found in COMMANDS")	2✔
204
205	return functools.partial(self._command, name)	2✔
206
207	def frame_received(self, data: bytes) -> None:	2✔
208	"""Handle a received EZSP frame
209
210	The protocol has taken care of UART specific framing etc, so we should
211	just have EZSP application stuff here, with all escaping/stuffing and
212	data randomization removed.
213	"""
214	self._protocol(data)	2✔
215
216	def add_callback(self, cb):	2✔
217	id_ = hash(cb)	2✔
218	while id_ in self._callbacks:	2✔
219	id_ += 1	2✔
220	self._callbacks[id_] = cb	2✔
221	return id_	2✔
222
223	def remove_callback(self, id_):	2✔
224	return self._callbacks.pop(id_)	2✔
225
226	def handle_callback(self, *args):	2✔
227	for callback_id, handler in self._callbacks.items():	2✔
228	try:	2✔
229	handler(*args)	2✔
230	except Exception as e:	2✔
231	LOGGER.exception("Exception running handler", exc_info=e)	2✔
232
233	def update_policies(self, zigpy_config: dict) -> Coroutine:	2✔
234	"""Set up the policies for what the NCP should do."""
235	return self._protocol.update_policies(zigpy_config)	×
236
237	def start_ezsp(self):	2✔
238	"""Mark EZSP as running."""
239	self._ezsp_event.set()	2✔
240
241	def stop_ezsp(self):	2✔
242	"""Mark EZSP stopped."""
243	self._ezsp_event.clear()	2✔
244
245	@property	2✔
246	def is_ezsp_running(self):	1✔
247	"""Return True if EZSP is running."""
248	return self._ezsp_event.is_set()	2✔
249
250	@property	2✔
251	def ezsp_version(self):	1✔
252	"""Return protocol version."""
253	return self._ezsp_version	2✔
254
255	@property	2✔
256	def types(self):	1✔
257	"""Return EZSP types for this specific version."""
258	return self._protocol.types	×

zigpy / bellows / 989

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous