Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 13 additions & 14 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -1,43 +1,42 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
rev: v6.0.0
hooks:
- id: debug-statements

- repo: https://github.com/psf/black
rev: 23.10.1
rev: 26.3.1
hooks:
- id: black

- repo: https://github.com/PyCQA/flake8
rev: 6.1.0
rev: 7.3.0
hooks:
- id: flake8
entry: pflake8
additional_dependencies:
- pyproject-flake8==6.1.0
- flake8-bugbear==23.1.20
- flake8-comprehensions==3.10.1
- flake8_2020==1.7.0
additional_dependencies:
- Flake8-pyproject
- flake8-bugbear==25.11.29
- flake8-comprehensions==3.17.0
- flake8_2020==1.8.1
- mccabe==0.7.0
- pycodestyle==2.11.1
- pyflakes==3.1.0
- pycodestyle==2.14.0
- pyflakes==3.4.0

- repo: https://github.com/PyCQA/isort
rev: 5.12.0
rev: 8.0.1
hooks:
- id: isort

- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.6.1
rev: v1.20.1
hooks:
- id: mypy
additional_dependencies:
- zigpy
- types-setuptools

- repo: https://github.com/asottile/pyupgrade
rev: v3.15.0
rev: v3.21.2
hooks:
- id: pyupgrade

Expand Down
5 changes: 2 additions & 3 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,12 @@ authors = [
]
readme = "README.md"
license = {text = "GPL-3.0"}
requires-python = ">=3.8"
requires-python = ">=3.11"
dependencies = [
"voluptuous",
"zigpy>=0.70.0",
"zigpy>=1.3.0",
"pyusb>=1.1.0",
"gpiozero",
'async-timeout; python_version<"3.11"',
]

[tool.setuptools.packages.find]
Expand Down
3 changes: 1 addition & 2 deletions tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
from unittest.mock import AsyncMock, MagicMock, patch, sentinel

import pytest
import serial_asyncio_fast
import zigpy.config as config

from zigpy_zigate import api as zigate_api
Expand Down Expand Up @@ -32,7 +31,7 @@ async def mock_conn(loop, protocol_factory, **kwargs):
loop.call_soon(protocol.connection_made, None)
return None, protocol

monkeypatch.setattr(serial_asyncio_fast, "create_serial_connection", mock_conn)
monkeypatch.setattr(zigpy_zigate.uart, "create_serial_connection", mock_conn)

await api.connect()

Expand Down
2 changes: 1 addition & 1 deletion tests/test_application.py
Original file line number Diff line number Diff line change
Expand Up @@ -229,7 +229,7 @@ async def test_energy_scanning(app, caplog):
channels=zigpy_t.Channels.ALL_CHANNELS, duration_exp=2, count=5
)

assert scan_results == {c: 0 for c in zigpy_t.Channels.ALL_CHANNELS}
assert scan_results == dict.fromkeys(zigpy_t.Channels.ALL_CHANNELS, 0)

# We never send a request when scanning
assert len(app._api.raw_aps_data_request.mock_calls) == 0
Expand Down
6 changes: 3 additions & 3 deletions tests/test_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@


def test_deserialize():
extra = b"\xBE\xEF"
extra = b"\xbe\xef"
data = b"\x00\x01\x00\x02"
schema = RESPONSES[0x8000]
result, rest = t.deserialize(data + extra, schema)
Expand All @@ -15,7 +15,7 @@ def test_deserialize():
assert result[2] == 0x0002
assert result[3] == extra

extra = b"\xBE\xEF"
extra = b"\xbe\xef"
data = b"\x00\x00\x01\x00\x01\x01\x01\x02\x12\x34\x02\xab\xcd\x01\x00"
schema = RESPONSES[0x8002]
result, rest = t.deserialize(data + extra, schema)
Expand All @@ -26,7 +26,7 @@ def test_deserialize():
assert result[4] == 0x01
assert result[5] == t.Address(address_mode=t.AddressMode.NWK, address=t.NWK(0x1234))
assert result[6] == t.Address(address_mode=t.AddressMode.NWK, address=t.NWK(0xABCD))
assert result[7] == b"\x01\x00\xBE\xEF"
assert result[7] == b"\x01\x00\xbe\xef"
assert rest == b""

data = b"\x00\x01\x01\x02\x12\x34\xff"
Expand Down
53 changes: 31 additions & 22 deletions tests/test_uart.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,7 @@

import gpiozero
import pytest
import serial.tools.list_ports
import serial_asyncio_fast
from serialx import SerialPortInfo
import zigpy.config

from zigpy_zigate import common, uart
Expand Down Expand Up @@ -32,7 +31,7 @@ async def mock_conn(loop, protocol_factory, url, **kwargs):
assert url.startswith("/") is True
return None, protocol

monkeypatch.setattr(serial_asyncio_fast, "create_serial_connection", mock_conn)
monkeypatch.setattr(uart, "create_serial_connection", mock_conn)
monkeypatch.setattr(common, "set_pizigate_running_mode", AsyncMock())
DEVICE_CONFIG = zigpy.config.SCHEMA_DEVICE({zigpy.config.CONF_DEVICE_PATH: port})

Expand Down Expand Up @@ -81,7 +80,7 @@ def test_data_received_incomplete_frame(gw):


def test_data_received_runt_frame(gw):
data = b"\x02\x44\xC0"
data = b"\x02\x44\xc0"
gw.data_received(data)
assert gw._api.data_received.call_count == 0

Expand Down Expand Up @@ -137,32 +136,42 @@ def test_is_not_pizigate():


def test_is_zigatedin(monkeypatch):
def mock_grep(*args, **kwargs):
device = MagicMock()
device.description = "ZiGate"
device.manufacturer = "FTDI"
return iter([device])

monkeypatch.setattr(serial.tools.list_ports, "grep", mock_grep)
port = "/dev/ttyUSB1"
r = common.is_zigate_din(port)
assert r is True
device = SerialPortInfo(
device=port,
resolved_device=port,
vid=None,
pid=None,
serial_number=None,
manufacturer="FTDI",
product="ZiGate",
bcd_device=None,
interface_description=None,
interface_num=None,
)
monkeypatch.setattr(common, "list_serial_ports", lambda: [device])
assert common.is_zigate_din(port) is True


@pytest.mark.parametrize(
"port",
("/dev/ttyUSB1", "/dev/ttyAMA0", "/dev/serial0"),
)
def test_is_not_zigatedin(port, monkeypatch):
def mock_grep(*args, **kwargs):
device = MagicMock()
device.description = "Other"
device.manufacturer = "FTDI"
return iter([device])

monkeypatch.setattr(serial.tools.list_ports, "grep", mock_grep)
r = common.is_zigate_din(port)
assert r is False
device = SerialPortInfo(
device=port,
resolved_device=port,
vid=None,
pid=None,
serial_number=None,
manufacturer="FTDI",
product="Other",
bcd_device=None,
interface_description=None,
interface_num=None,
)
monkeypatch.setattr(common, "list_serial_ports", lambda: [device])
assert common.is_zigate_din(port) is False


def test_is_zigate_wifi():
Expand Down
20 changes: 9 additions & 11 deletions zigpy_zigate/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,7 @@
import time

from gpiozero import OutputDevice
import serial
import serial.tools.list_ports
from serialx import SerialException, list_serial_ports

LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -35,18 +34,19 @@ def __init__(

def discover_port():
"""discover zigate port"""
devices = list(serial.tools.list_ports.grep("ZiGate"))
ports = list_serial_ports()
devices = [p for p in ports if p.product and "ZiGate" in p.product]
if devices:
port = devices[0].device
LOGGER.info("ZiGate found at %s", port)
else:
devices = list(serial.tools.list_ports.grep("067b:2303|CP2102"))
devices = [p for p in ports if re.search(r"067b:2303|CP2102", p.product or "")]
Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think discover_port is used anywhere so this shouldn't really matter. We're going to be deprecating this library anyways, I'd rather not spend much time fixing dead code.

if devices:
port = devices[0].device
LOGGER.info("ZiGate probably found at %s", port)
else:
LOGGER.error("Unable to find ZiGate using auto mode")
raise serial.SerialException("Unable to find Zigate using auto mode")
raise SerialException("Unable to find Zigate using auto mode")
return port


Expand All @@ -63,12 +63,10 @@ def is_zigate_din(port):
"""detect zigate din"""
port = os.path.realpath(port)
if re.match(r"/dev/ttyUSB\d+", port):
try:
device = next(serial.tools.list_ports.grep(port))
# Suppose zigate din /dev/ttyUSBx
return device.description == "ZiGate" and device.manufacturer == "FTDI"
except StopIteration:
pass
for device in list_serial_ports():
if device.device == port:
# Suppose zigate din /dev/ttyUSBx
return device.product == "ZiGate" and device.manufacturer == "FTDI"
return False


Expand Down
2 changes: 1 addition & 1 deletion zigpy_zigate/tools/cli.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""
Simple CLI ZiGate tool
Simple CLI ZiGate tool
"""

import argparse
Expand Down
6 changes: 3 additions & 3 deletions zigpy_zigate/uart.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,14 @@
from typing import Any

import zigpy.config
import zigpy.serial
from zigpy.serial import SerialProtocol, create_serial_connection

from . import common as c

LOGGER = logging.getLogger(__name__)


class Gateway(zigpy.serial.SerialProtocol):
class Gateway(SerialProtocol):
START = b"\x01"
END = b"\x03"

Expand Down Expand Up @@ -126,7 +126,7 @@ async def connect(device_config: dict[str, Any], api, loop=None):
await c.async_set_zigatedin_running_mode()

protocol = Gateway(api)
_, protocol = await zigpy.serial.create_serial_connection(
_, protocol = await create_serial_connection(
loop,
lambda: protocol,
url=port,
Expand Down
2 changes: 1 addition & 1 deletion zigpy_zigate/zigbee/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ async def energy_scan(
"""Runs an energy detection scan and returns the per-channel scan results."""

LOGGER.warning("Coordinator does not support energy scanning")
return {c: 0 for c in channels}
return dict.fromkeys(channels, 0)

async def force_remove(self, dev):
await self._api.remove_device(self.state.node_info.ieee, dev.ieee)
Expand Down