diff --git a/examples/cmd.py b/examples/cmd.py new file mode 100755 index 0000000..317cee0 --- /dev/null +++ b/examples/cmd.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +import time +import sys + +from pa1010d import PA1010D + + +""" +Run raw commands against the PA1010D GPS and return the respones. + +Eg: + PMTK605 = Query Firmware Release Info + PMTK430 = Query Datum + PMTK414 = Query NMEA Output Frequency + PMTK400 = Query Update Rate + PMTK225,<1 or 0> = Enable/Disable PPS +""" + +def timeout(err=None, timeout=5.0): + if err is None: + err = "Timed out!" + t_start = time.time() + while time.time() - t_start < timeout: + yield + raise TimeoutError(err) + + +responses = { +} + +if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} ") + sys.exit() + +command = sys.argv[1] +response = responses.get(command, f"$PMTK{int(command[4:]) + 100}") + +gps = PA1010D() + +gps.update() + +gps.send_command(sys.argv[1]) + +if response: + print(f"Waiting for {response}...") + for t in timeout("Timed out waiting for command response."): + message = gps.read_sentence() + if message.startswith(response): + print(message) + break diff --git a/examples/pps.py b/examples/pps.py new file mode 100755 index 0000000..53f586b --- /dev/null +++ b/examples/pps.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +import time +import sys + +from pa1010d import PA1010D + + +if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} ") + sys.exit() + +gps = PA1010D() + +if sys.argv[1] == "on": + gps.send_command("PMTK255,1") +else: + gps.send_command("PMTK255,0") + +result = gps.update() + +print("OK" if result else "Uh oh!") diff --git a/library/pa1010d/__init__.py b/library/pa1010d/__init__.py index ed75844..b41e7a3 100644 --- a/library/pa1010d/__init__.py +++ b/library/pa1010d/__init__.py @@ -67,7 +67,7 @@ def _write_sentence(self, bytestring): """ for char_index in bytestring: - self._i2c.write_byte(self._i2c_addr, ord(char_index)) + self._i2c.write_byte(self._i2c_addr, char_index) def send_command(self, command, add_checksum=True): """Send a command string to the PA1010D. @@ -75,10 +75,13 @@ def send_command(self, command, add_checksum=True): If add_checksum is True (the default) a NMEA checksum will automatically be computed and added. """ + if type(command) is not bytes: + command = command.encode("ascii") + # TODO replace with pynmea2 functionality - if command.startswith("$"): + if command[0] == b"$": command = command[1:] - if command.endswith("*"): + if command[-1] == b"*": command = command[:-1] buf = bytearray() @@ -112,7 +115,7 @@ def read_sentence(self, timeout=5): # Should be a full \r\n since the GPS emits spurious newlines if buf[-2:] == [ord("\r"), ord("\n")]: # Remove line ending and spurious newlines from the sentence - return bytearray(buf).decode("ascii").strip().replace("\n","") + return bytearray(buf).decode("ascii").strip().replace("\n", "") raise TimeoutError("Timeout waiting for readline") diff --git a/library/tests/conftest.py b/library/tests/conftest.py index 8e1cb33..c119e3f 100644 --- a/library/tests/conftest.py +++ b/library/tests/conftest.py @@ -3,9 +3,28 @@ import sys +class SMBus: + def __init__(self, bus): + self.data = "$PMTK011,MTKGPS*08\r\n".encode("ascii") + self.ptr = 0 + + def read_byte_data(self, address, register): + if register == 0x00: + result = self.data[self.ptr] + self.ptr += 1 + self.ptr %= len(self.data) + return result + else: + return 0 + + def write_byte(self, address, data): + pass + + @pytest.fixture(scope='function', autouse=False) def smbus(): smbus = mock.MagicMock() + smbus.SMBus = SMBus sys.modules["smbus"] = smbus yield smbus del sys.modules["smbus"] diff --git a/library/tests/test_setup.py b/library/tests/test_setup.py index 37d3029..8e4e73e 100644 --- a/library/tests/test_setup.py +++ b/library/tests/test_setup.py @@ -3,3 +3,19 @@ def test_setup(smbus): gps = pa1010d.PA1010D() del gps + + +def test_send_command(smbus): + import pa1010d + + gps = pa1010d.PA1010D() + gps.send_command("$TEST") + gps.send_command("$TEST*") + gps.send_command("$TEST*".encode("ascii")) + + +def test_recv_command(smbus): + import pa1010d + + gps = pa1010d.PA1010D() + assert gps.update() is True