mirror of
https://github.com/meshtastic/python.git
synced 2026-01-17 12:17:55 -05:00
wip; testing gpio read not working for me on real device; worked on gpio watch
This commit is contained in:
4
Makefile
4
Makefile
@@ -10,6 +10,10 @@ install:
|
||||
lint:
|
||||
pylint meshtastic
|
||||
|
||||
# show the slowest unit tests
|
||||
slow:
|
||||
pytest --durations=0
|
||||
|
||||
# run the coverage report and open results in a browser
|
||||
cov:
|
||||
pytest --cov-report html --cov=meshtastic
|
||||
|
||||
12
README.md
12
README.md
@@ -199,6 +199,8 @@ pytest -vv
|
||||
pytest
|
||||
# or (more verbosely)
|
||||
pytest -m unit
|
||||
# or
|
||||
make
|
||||
```
|
||||
|
||||
* To run just integration tests:
|
||||
@@ -246,4 +248,14 @@ pytest -m smokewifi meshtastic/tests/test_smoke_wifi.py::test_smokewifi_info
|
||||
pytest --cov=meshtastic
|
||||
# or if want html coverage report
|
||||
pytest --cov-report html --cov=meshtastic
|
||||
# or
|
||||
make cov
|
||||
```
|
||||
|
||||
* To see slowest unit tests, you can run:
|
||||
|
||||
```
|
||||
pytest --durations=0
|
||||
# or
|
||||
make slow
|
||||
```
|
||||
|
||||
@@ -273,6 +273,7 @@ def onConnected(interface):
|
||||
if args.gpio_rd:
|
||||
bitmask = int(args.gpio_rd, 16)
|
||||
print(f"Reading GPIO mask 0x{bitmask:x} from {args.dest}")
|
||||
interface.mask = bitmask
|
||||
rhc.readGPIOs(args.dest, bitmask, None)
|
||||
if not interface.noProto:
|
||||
# wait up to X seconds for a response
|
||||
@@ -284,8 +285,10 @@ def onConnected(interface):
|
||||
|
||||
if args.gpio_watch:
|
||||
bitmask = int(args.gpio_watch, 16)
|
||||
print(f"Watching GPIO mask 0x{bitmask:x} from {args.dest}")
|
||||
rhc.watchGPIOs(args.dest, bitmask)
|
||||
print(f"Watching GPIO mask 0x{bitmask:x} from {args.dest}. Press ctrl-c to exit")
|
||||
while True:
|
||||
rhc.watchGPIOs(args.dest, bitmask)
|
||||
time.sleep(1)
|
||||
|
||||
# handle settings
|
||||
if args.set:
|
||||
|
||||
@@ -53,7 +53,8 @@ class MeshInterface:
|
||||
self.nodesByNum = None
|
||||
self.configId = None
|
||||
self.defaultHopLimit = 3
|
||||
self.gotResponse = False
|
||||
self.gotResponse = False # used in gpio read
|
||||
self.mask = None # used in gpio read and gpio watch
|
||||
|
||||
def close(self):
|
||||
"""Shutdown this interface"""
|
||||
|
||||
@@ -11,7 +11,10 @@ def onGPIOreceive(packet, interface):
|
||||
"""
|
||||
logging.debug(f"packet:{packet} interface:{interface}")
|
||||
hw = packet["decoded"]["remotehw"]
|
||||
print(f'Received RemoteHardware typ={hw["typ"]}, gpio_value={hw["gpioValue"]}')
|
||||
gpioValue = hw["gpioValue"]
|
||||
print(f'mask:{interface.mask}')
|
||||
value = int(gpioValue) & int(interface.mask)
|
||||
print(f'Received RemoteHardware typ={hw["typ"]}, gpio_value={gpioValue} value={value}')
|
||||
interface.gotResponse = True
|
||||
|
||||
|
||||
@@ -73,4 +76,5 @@ class RemoteHardwareClient:
|
||||
r = remote_hardware_pb2.HardwareMessage()
|
||||
r.typ = remote_hardware_pb2.HardwareMessage.Type.WATCH_GPIOS
|
||||
r.gpio_mask = mask
|
||||
self.iface.mask = mask
|
||||
return self._sendHardware(nodeid, r)
|
||||
|
||||
@@ -23,7 +23,7 @@ def test_RemoteHardwareClient():
|
||||
def test_onGPIOreceive(capsys):
|
||||
"""Test onGPIOreceive"""
|
||||
iface = MagicMock(autospec=SerialInterface)
|
||||
packet = {'decoded': {'remotehw': {'typ': 'foo', 'gpioValue': 'bar' }}}
|
||||
packet = {'decoded': {'remotehw': {'typ': 'foo', 'gpioValue': '4096' }}}
|
||||
onGPIOreceive(packet, iface)
|
||||
out, err = capsys.readouterr()
|
||||
assert re.search(r'Received RemoteHardware', out)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[pytest]
|
||||
|
||||
addopts = -m "not smoke1 and not smoke2 and not smokewifi and not examples"
|
||||
addopts = -m "not int and not smoke1 and not smoke2 and not smokewifi and not examples"
|
||||
|
||||
markers =
|
||||
unit: marks tests as unit tests
|
||||
|
||||
Reference in New Issue
Block a user