Merge pull request #140 from semuconsulting/RELEASE-CANDIDATE-1.2.38

Release candidate 1.2.38

.vscode/settings.json

@@ -4,5 +4,5 @@
     "editor.formatOnSave": true,
     "modulename": "${workspaceFolderBasename}",
     "distname": "${workspaceFolderBasename}",
-    "moduleversion": "1.2.37",
+    "moduleversion": "1.2.38",
 }

README.md

@@ -95,7 +95,8 @@ conda install -c conda-forge pyubx2
 | POLL (0x02) | query input *to* the receiver            | `ubxtypes_poll.py` |
 
 If you're simply streaming and/or parsing the *output* of a UBX receiver, the mode is implicitly GET. If you want to create
-or parse an *input* (command or query) message, you must set the mode parameter to SET or POLL.
+or parse an *input* (command or query) message, you must set the mode parameter to SET or POLL. If the parser mode is set to
+0x03 (SETPOLL), `pyubx2` will automatically determine the applicable input mode (SET or POLL) based on the message payload.
 
 ---
 ## <a name="reading">Reading (Streaming)</a>
@@ -116,7 +117,7 @@ The constructor accepts the following optional keyword arguments:
 * `quitonerror`: 0 = ignore errors, 1 = log errors and continue (default), 2 = (re)raise errors and terminate
 * `validate`: VALCKSUM (0x01) = validate checksum (default), VALNONE (0x00) = ignore invalid checksum or length
 * `parsebitfield`: 1 = parse bitfields ('X' type properties) as individual bit flags, where defined (default), 0 = leave bitfields as byte sequences
-* `msgmode`: 0 = GET (default), 1 = SET, 2 = POLL
+* `msgmode`: 0 = GET (default), 1 = SET, 2 = POLL, 3 = SETPOLL (automatically determine SET or POLL input mode)
 
 Example -  Serial input. This example will output both UBX and NMEA messages:
 ```python
@@ -159,7 +160,7 @@ The `parse()` method accepts the following optional keyword arguments:
 
 * `validate`: VALCKSUM (0x01) = validate checksum (default), VALNONE (0x00) = ignore invalid checksum or length
 * `parsebitfield`: 1 = parse bitfields as individual bit flags, where defined (default), 0 = leave bitfields as byte sequences
-* `msgmode`: 0 = GET (default), 1 = SET, 2 = POLL
+* `msgmode`: 0 = GET (default), 1 = SET, 2 = POLL, 3 = SETPOLL (automatically determine SET or POLL input mode)
 
 Example - output (GET) message:
 ```python

RELEASE_NOTES.md

@@ -1,6 +1,14 @@
 # pyubx2 Release Notes
 
-### RELEASE CANDIDATE 1.2.37
+### RELEASE CANDIDATE 1.2.38
+
+CHANGES:
+
+1. Add val2sphp helper method to convert high precision (9dp) coordinate to separate standard and high precision components, as required by some CFG and NAV messages.
+1. Add utc2itow helper method to convert utc datetime to GPS week number and time of week.
+1. Add getinputmode helper to determinate mode of input UBX message (SET or POLL). Add new UBXReader msgmode of SETPOLL (0x03), which will automatically determine input mode.
+
+### RELEASE 1.2.37
 
 CHANGES:
 

examples/f9p_basestation.py

@@ -17,13 +17,13 @@
 :license: BSD 3-Clause
 """
 
-from math import trunc
 from serial import Serial
-from pyubx2 import UBXMessage
+
+from pyubx2 import UBXMessage, val2sphp
 
 TMODE_SVIN = 1
 TMODE_FIXED = 2
-
+SHOW_PRESET = True  # hide or show PyGPSClient preset string
 
 def send_msg(serial_out: Serial, ubx: UBXMessage):
     """
@@ -105,17 +105,8 @@ def config_fixed(acc_limit: int, lat: float, lon: float, height: float) -> UBXMe
     layers = 1
     transaction = 0
     acc_limit = int(round(acc_limit / 0.1, 0))
-
-    # separate standard and high precision parts of lat / lon
-    # and apply scaling factors
-    lat_7dp = trunc(lat * 1e7) / 1e7
-    lat_hp = lat - lat_7dp
-    lat = int(round(lat_7dp / 1e-7, 0))
-    lat_hp = int(round(lat_hp / 1e-9, 0))
-    lon_7dp = trunc(lon * 1e7) / 1e7
-    lon_hp = lon - lon_7dp
-    lon = int(round(lon_7dp / 1e-7, 0))
-    lon_hp = int(round(lon_hp / 1e-9, 0))
+    lats, lath = val2sphp(lat)
+    lons, lonh = val2sphp(lon)
 
     height = int(height)
     cfg_data = [
@@ -124,10 +115,10 @@ def config_fixed(acc_limit: int, lat: float, lon: float, height: float) -> UBXMe
         ("CFG_TMODE_FIXED_POS_ACC", acc_limit),
         ("CFG_TMODE_HEIGHT_HP", 0),
         ("CFG_TMODE_HEIGHT", height),
-        ("CFG_TMODE_LAT", lat),
-        ("CFG_TMODE_LAT_HP", lat_hp),
-        ("CFG_TMODE_LON", lon),
-        ("CFG_TMODE_LON_HP", lon_hp),
+        ("CFG_TMODE_LAT", lats),
+        ("CFG_TMODE_LAT_HP", lath),
+        ("CFG_TMODE_LON", lons),
+        ("CFG_TMODE_LON_HP", lonh),
     ]
 
     ubx = UBXMessage.config_set(layers, transaction, cfg_data)
@@ -147,21 +138,22 @@ def config_fixed(acc_limit: int, lat: float, lon: float, height: float) -> UBXMe
     PORT_TYPE = "USB"  # choose from "USB", "UART1", "UART2"
     BAUD = 38400
     TIMEOUT = 5
-    SHOW_PRESET = True  # hide or show PyGPSClient preset string
 
-    TMODE = TMODE_SVIN  # "TMODE_SVIN" or 1 = Survey-In, "TMODE_FIXED" or 2 = Fixed
+
+    TMODE = TMODE_FIXED  # "TMODE_SVIN" or 1 = Survey-In, "TMODE_FIXED" or 2 = Fixed
     ACC_LIMIT = 200  # accuracy in mm
 
-    # only used if TMODE = 1 ...
+    # only used if TMODE = SVIN ...
     SVIN_MIN_DUR = 90  # seconds
 
-    # only used if TMODE = 2 ...
-    ARP_LAT = 37.012345678
-    ARP_LON = -115.012345678
+    # only used if TMODE = FIXED ...
+    ARP_LAT = 12.123456789
+    ARP_LON = -115.987654321
     ARP_HEIGHT = 137000  # cm
 
     print(f"Configuring receiver on {PORT} @ {BAUD:,} baud.\n")
     with Serial(PORT, BAUD, timeout=TIMEOUT) as stream:
+
         # configure RTCM3 outputs
         msg = config_rtcm(PORT_TYPE)
         send_msg(stream, msg)

pyproject.toml

@@ -7,7 +7,7 @@ name = "pyubx2"
 authors = [{ name = "semuadmin", email = "[email protected]" }]
 maintainers = [{ name = "semuadmin", email = "[email protected]" }]
 description = "UBX protocol parser and generator"
-version = "1.2.37"
+version = "1.2.38"
 license = { file = "LICENSE" }
 readme = "README.md"
 requires-python = ">=3.8"
@@ -87,7 +87,7 @@ disable = """
 
 [tool.pytest.ini_options]
 minversion = "7.0"
-addopts = "--cov --cov-report term-missing --cov-fail-under 95"
+addopts = "--cov --cov-report html --cov-fail-under 95"
 pythonpath = ["src"]
 testpaths = ["tests"]
 

src/pyubx2/_version.py

@@ -8,4 +8,4 @@
 :license: BSD 3-Clause
 """
 
-__version__ = "1.2.37"
+__version__ = "1.2.38"

src/pyubx2/ubxhelpers.py

@@ -13,14 +13,19 @@
 
 import struct
 from datetime import datetime, timedelta
-from math import cos, pi, sin
+from math import cos, pi, sin, trunc
 
 from pynmeagps.nmeatypes_core import NMEA_HDR
 
 import pyubx2.exceptions as ube
 import pyubx2.ubxtypes_configdb as ubcdb
 import pyubx2.ubxtypes_core as ubt
-from pyubx2.ubxtypes_core import GNSSLIST, UBX_HDR
+from pyubx2.ubxtypes_core import POLL, SET, UBX_HDR
+from pyubx2.ubxtypes_decodes import FIXTYPE, GNSSLIST
+
+EPOCH0 = datetime(1980, 1, 6)  # EPOCH start date
+LEAPOFFSET = 18  # leap year offset in seconds, valid as from 1/1/2017
+SIW = 604800  # seconds in week = 3600*24*7
 
 
 def att2idx(att: str) -> int:
@@ -123,18 +128,33 @@ def attsiz(att: str) -> int:
 def itow2utc(itow: int) -> datetime.time:
     """
     Convert GPS Time Of Week to UTC time
-    (UTC = GPS - 18 seconds; correct as from 1/1/2017).
 
-    :param int itow: GPS Time Of Week
+    :param int itow: GPS Time Of Week in milliseconds
     :return: UTC time hh.mm.ss
     :rtype: datetime.time
 
     """
 
-    utc = datetime(1980, 1, 6) + timedelta(seconds=(itow / 1000) - 18)
+    utc = EPOCH0 + timedelta(seconds=(itow / 1000) - LEAPOFFSET)
     return utc.time()
 
 
+def utc2itow(utc: datetime) -> tuple:
+    """
+    Convert UTC datetime to GPS Week Number, Time Of Week
+
+    :param datetime utc: datetime
+    :return: GPS Week Number, Time of Week in milliseconds
+    :rtype: tuple
+
+    """
+
+    wno = int((utc - EPOCH0).total_seconds() / SIW)
+    sow = EPOCH0 + timedelta(seconds=wno * SIW)
+    itow = int(((utc - sow).total_seconds() + LEAPOFFSET) * 1000)
+    return wno, itow
+
+
 def gpsfix2str(fix: int) -> str:
     """
     Convert GPS fix integer to descriptive string.
@@ -145,19 +165,10 @@ def gpsfix2str(fix: int) -> str:
 
     """
 
-    if fix == 5:
-        fixs = "TIME ONLY"
-    elif fix == 4:
-        fixs = "GPS + DR"
-    elif fix == 3:
-        fixs = "3D"
-    elif fix == 2:
-        fixs = "2D"
-    elif fix == 1:
-        fixs = "DR"
-    else:
-        fixs = "NO FIX"
-    return fixs
+    try:
+        return FIXTYPE[fix]
+    except KeyError:
+        return str(fix)
 
 
 def dop2str(dop: float) -> str:
@@ -498,3 +509,50 @@ def escapeall(val: bytes) -> str:
     """
 
     return "b'{}'".format("".join(f"\\x{b:02x}" for b in val))
+
+
+def val2sphp(val: float, scale: float = 1e-7) -> tuple:
+    """
+    Convert a float value into separate standard and high precisions components,
+    multiplied by a scaling factor to render them as integers, as required by some
+    CFG and NAV messages.
+
+    e.g. 48.123456789 becomes (481234567, 89)
+
+    :param float val: value as float
+    :param float scale: scaling factor e.g. 1e-7
+    :return: tuple of (standard precision, high precision)
+    :rtype: tuple
+    """
+
+    val = val / scale
+    val_sp = trunc(val)
+    val_hp = round((val - val_sp) * 100)
+    return val_sp, val_hp
+
+
+def getinputmode(data: bytes) -> int:
+    """
+    Return input message mode (SET or POLL).
+
+    :param bytes data: raw UBX input message
+    :return: message mode (1 = SET, 2 = POLL)
+    :rtype: int
+    """
+
+    if (
+        len(data) == 8
+        or data[2:4] == b"\x06\x8b"  # CFG-VALGET
+        or (
+            data[2:4]
+            in (
+                b"\x06\x01",
+                b"\x06\x02",
+                b"\x06\x03",
+                b"\x06\x31",
+            )  # CFG-INF, CFG-MSG, CFG-PRT, CFG-TP5
+            and len(data) <= 10
+        )
+    ):
+        return POLL
+    return SET

src/pyubx2/ubxmessage.py

@@ -74,8 +74,8 @@ def __init__(
         self._parsebf = parsebitfield  # parsing bitfields Y/N?
         self._scaling = scaling  # apply scale factors Y/N?
 
-        if msgmode not in (0, 1, 2):
-            raise ube.UBXMessageError(f"Invalid msgmode {msgmode} - must be 0, 1 or 2.")
+        if msgmode not in (ubt.GET, ubt.SET, ubt.POLL):
+            raise ube.UBXMessageError(f"Invalid msgmode {msgmode} - must be 0, 1 or 2")
 
         # accommodate different formats of msgClass and msgID
         if isinstance(ubxClass, str) and isinstance(

src/pyubx2/ubxreader.py

@@ -9,7 +9,9 @@
 
 'protfilter' governs which protocols (NMEA, UBX or RTCM3) are processed
 'quitonerror' governs how errors are handled
-'msgmode' indicates the type of UBX datastream (input GET, output SET, query POLL)
+'msgmode' indicates the type of UBX datastream (output GET, input SET, query POLL)
+
+If msgmode set to SETPOLL, input mode will be automatically detected by parser.
 
 Created on 2 Oct 2020
 
@@ -32,14 +34,17 @@
     UBXTypeError,
 )
 from pyubx2.socket_stream import SocketStream
-from pyubx2.ubxhelpers import bytes2val, calc_checksum, val2bytes
+from pyubx2.ubxhelpers import bytes2val, calc_checksum, getinputmode, val2bytes
 from pyubx2.ubxmessage import UBXMessage
 from pyubx2.ubxtypes_core import (
     ERR_LOG,
     ERR_RAISE,
     GET,
     NMEA_PROTOCOL,
+    POLL,
     RTCM3_PROTOCOL,
+    SET,
+    SETPOLL,
     U2,
     UBX_HDR,
     UBX_PROTOCOL,
@@ -69,7 +74,7 @@ def __init__(
         """Constructor.
 
         :param datastream stream: input data stream
-        :param int msgmode: 0=GET, 1=SET, 2=POLL (0)
+        :param int msgmode: 0=GET, 1=SET, 2=POLL, 3=SETPOLL (0)
         :param int validate: 0 = ignore invalid checksum, 1 = validate checksum (1)
         :param int protfilter: protocol filter 1 = NMEA, 2 = UBX, 4 = RTCM3 (3)
         :param int quitonerror: 0 = ignore errors,  1 = log continue, 2 = (re)raise (1)
@@ -97,9 +102,9 @@ def __init__(
         self._msgmode = msgmode
         self._parsing = parsing
 
-        if self._msgmode not in (0, 1, 2):
+        if self._msgmode not in (GET, SET, POLL, SETPOLL):
             raise UBXStreamError(
-                f"Invalid stream mode {self._msgmode} - must be 0, 1 or 2"
+                f"Invalid stream mode {self._msgmode} - must be 0, 1, 2 or 3"
             )
 
     def __iter__(self):
@@ -374,8 +379,10 @@ def parse(
         """
         # pylint: disable=too-many-arguments
 
-        if msgmode not in (0, 1, 2):
-            raise UBXParseError(f"Invalid message mode {msgmode} - must be 0, 1 or 2")
+        if msgmode not in (GET, SET, POLL, SETPOLL):
+            raise UBXParseError(
+                f"Invalid message mode {msgmode} - must be 0, 1, 2 or 3"
+            )
 
         lenm = len(message)
         hdr = message[0:2]
@@ -410,6 +417,9 @@ def parse(
                     (f"Message checksum {ckm}" f" invalid - should be {ckv}")
                 )
         try:
+            # if input message (SET or POLL), determine mode automatically
+            if msgmode == SETPOLL:
+                msgmode = getinputmode(message)  # returns SET or POLL
             if payload is None:
                 return UBXMessage(clsid, msgid, msgmode)
             return UBXMessage(