diff --git a/README.md b/README.md index 0b73eddd..455cbb35 100644 --- a/README.md +++ b/README.md @@ -399,15 +399,6 @@ By default, the server/caster binds to the host address '0.0.0.0' (IPv4) or '::' ยน Unicore and Septentrio devices do not appear to support an SVIN status output message, so Survey-in % completion is not available. ---- -## SPARTN Client Facilities - - ### NB: As of October 2025, u-blox have discontinued both their L-Band and MQTT encrypted SPARTN correction services, so the SPARTN Client functionality is effectively redundant and may be removed in a subsequent version of PyGPSClient. - - The SPARTN MQTT and L-Band configuration panels are now disabled by default, though the L-Band panel can in theory still be used for other generic L-Band modem configuration purposes and can be re-enabled by setting json configuration parameter `lband_enabled_b` to `1`. - -Please refer to [SPARTN.md](https://github.com/semuconsulting/PyGPSClient/blob/master/SPARTN.md) for instructions. - --- ## GPX Track Viewer diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 52dfb0d9..93030812 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,5 +1,17 @@ # PyGPSClient Release Notes +### RELEASE 1.7.2 + +FIXES: + +1. Various minor bug fixes. + - garbled configuration loaded message on startup. + - thread error when database logging started. + +CHANGES: + +1. [SPARTN Configuration Panel](https://github.com/semuconsulting/PyGPSClient/blob/master/SPARTN.md) removed. Recommend resaving any existing PyGPSClient json configuration files as several configuration settings referencing this panel will now be redundant. + ### RELEASE 1.7.1 FIXES: diff --git a/SPARTN.md b/SPARTN.md deleted file mode 100644 index 09c1ee19..00000000 --- a/SPARTN.md +++ /dev/null @@ -1,97 +0,0 @@ - -# PyGPSClient SPARTN Configuration Dialog - ---- -## SPARTN Client Facilities - - ### NB: As of October 2025, u-blox have discontinued both their L-Band and MQTT encrypted SPARTN correction services, so the SPARTN Client functionality is effectively redundant and may be removed in a subsequent version of PyGPSClient. - - The SPARTN MQTT and L-Band configuration panels are now disabled by default, though the L-Band panel can in theory still be used for other generic L-Band modem configuration purposes and can be re-enabled by setting json configuration parameter `lband_enabled_b` to `1`. - -![spartn config widget screenshot](https://github.com/semuconsulting/PyGPSClient/blob/master/images/spartnconfig_widget.png?raw=true) - -The SPARTN Configuration utility allows users to receive and process SPARTN RTK Correction data from an IP or L-Band source to achieve cm level location accuracy. It provides three independent configuration sections, one for IP Correction (MQTT), one for L-Band Correction (e.g. NEO-D9S) and a third for the GNSS receiver (e.g. ZED-F9P). - -The facility can be accessed by clicking ![SPARTN Client button](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/iconmonstr-antenna-3-24.png?raw=true) or selecting Menu..Options..SPARTN Configuration Dialog. - -**Pre-Requisites:** - -**NB:** SPARTN data from proprietary subscription services like Thingstream PointPerfect (MQTT or L-Band) is encrypted. While it is not necessary for PyGPSClient to decrypt the data prior to sending it to the GNSS receiver (*the receiver's own firmware will perform the necessary decryption provided the relevant keys have been uploaded*), if you wish to see the decrypted data in the PyGPSClient console it will be necessary to provide current decryption keys via the `spartndecode_b` and `spartnkey_s` settings in the json configuration file. Note that these keys are typically only valid for a 4 week period and will require regular updating. See [pyspartn Encrypted Payloads](https://github.com/semuconsulting/pyspartn#encrypted-payloads) for further details. - -1. IP Correction (MQTT Client): - - - Internet access - - Subscription to a suitable MQTT SPARTN location service e.g. u-blox / Thingstream PointPerfect IP or L-band, which should provide the following details: - - Server URL e.g. `pp.services.u-blox.com` - - Client ID (which can be stored in the `"mqttclientid_s":` json configuration file setting or via environment variable `MQTTCLIENTID`) - - Client Certificate (`*.crt`) and Client Key (`*.pem`) files required to access the SPARTN service via an encrypted HTTPS connection. These files can be downloaded from the Thingstream..PointPerfect..Location Things..Credentials web page. If these are placed in the user's HOME directory using the location service's standard naming convention, PyGPSClient will find them automatically. - - Region code - select from `us`, `eu`, `jp`, `kr` or `au`. - - Source - select from either `IP` or `L-Band` (*NB: the 'L-Band' MQTT mode provides decryption keys, Assist Now (ephemerides) data and L-Band frequency information, but the correction data itself arrives via the L-Band receiver below*). - - A list of published topics. These typically include: - - `/pp/ip/region` - binary SPARTN correction data (SPARTN-1X-HPAC* / OCB* / GAD*) for the selected region, for IP sources only. - - `/pp/ubx/mga` - UBX MGA AssistNow ephemera data for each constellation. - - `/pp/ubx/0236/ip` or `/pp/ubx/0236/Lb` - UBX RXM-SPARTNKEY messages containing the IP or L-band decryption keys to be uploaded to the GNSS receiver. - - `/pp/frequencies/Lb` - json message containing each region's L-band transmission frequency - currently `us` or `eu` (this is automatically enabled when `L-Band` is selected). - -2. L-BAND Correction (D9* Receiver): - - - SPARTN L-Band correction receiver e.g. u-blox NEO-D9S. - - [Suitable Inmarsat L-band antenna](https://www.amazon.com/RTL-SDR-Blog-1525-1637-Inmarsat-Iridium/dp/B07WGWZS1D) and good satellite reception on regional frequency (NB: standard GNSS antenna may not be suitable). - - Subscription to L-Band location service e.g. u-blox / Thingstream PointPerfect, which should provide the following details: - - L-Band frequency (*also available via `/pp/frequencies/Lb` MQTT topic*) - - Search window - - Use Service ID? - - Use Descrambling? - - Use Prescrambling? - - Service ID - - Data Rate - - Descrambler Init - - Unique Word - -3. GNSS Receiver: - - - SPARTN-compatible GNSS receiver e.g. u-blox ZED-F9P - - Subscription to either IP and/or L-Band location service(s) e.g. e.g. u-blox / Thingstream PointPerfect, which should provide the following details: - - Current and Next IP or L-band decryption Keys in hexadecimal format. These allow the receiver to decrypt the SPARTN message payloads. - - Valid From dates in YYYYMMDD format (keys normally valid for 4 week period). - -**Instructions:** - -### IP Correction Configuration (MQTT) - -1. Enter your MQTT Client ID (or set it up beforehand as *.json configuration setting `"mqttclientid_s":` or via environment variable `MQTTCLIENTID`). -1. Select the path to the MQTT `*.crt` and `*.pem` files provided by the location service (PyGPSClient will use the user's HOME directory by default). -1. Select the required region and subscription mode (`IP` or `L-Band`). -1. Select the required topics: - - IP - this is the raw SPARTN correction data as SPARTN-1X-HPAC* / OCB* / GAD* messages (required for IP; must be unchecked for L-Band). - - Assist - this is Assist Now data as UBX MGA-* messages. - - Key - this is the SPARTN IP or L-band decryption keys as a UBX RXM-SPARTNKEY message. -1. To connect to the MQTT server, click ![connect icon](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/ethernet-1-24.png?raw=true). To disconnect, click ![disconnect icon](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/iconmonstr-media-control-50-24.png?raw=true). - -### L-Band Correction Configuration (D9*) - -1. **NOTE** This panel is only available if json configuration setting `lband_enabled_b` is set to `1`. -1. To connect to the Correction receiver, select the receiver's port from the SPARTN dialog's Serial Port listbox and click ![connect icon](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/usbport-1-24.png?raw=true). To disconnect, click ![disconnect icon](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/iconmonstr-media-control-50-24.png?raw=true). -1. Select the required Output Port - this is the port used to connect the Correction receiver to the GNSS receiver e.g. UART2 or I2C. -1. If both Correction and GNSS receivers are connected to the same PyGPSClient workstation (e.g. via separate USB ports), it is possible to run the utility in Output Port = 'Passthough' mode, whereby the output data from the Correction receiver (UBX `RXM-PMP` messages) will be automatically passed through to the GNSS receiver by PyGPSClient, without the need to connect the two externally. -1. To enable INF-DEBUG messages, which give diagnostic information about current L-Band reception, click 'Enable Debug?'. -1. To save the configuration to the device's persistent storage (Battery-backed RAM, Flash or EEPROM), click 'Save Config?'. This only has to be done once for a given region. -1. Once connected, click ![send button](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/iconmonstr-arrow-12-24.png?raw=true) to upload the configuration. The utility will send the relevant configuration command(s) to the Correction receiver and poll for an acknowledgement. - -### GNSS Receiver Configuration (F9*) - -1. Connect to the GNSS receiver first by clicking ![connect icon](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/usbport-1-24.png?raw=true) on the main PyGPSClient settings panel. -1. If you are subscribing to the MQTT SPARTNKEY (`/pp/ubx/0236/ip` or `/pp/ubx/0236/Lb`) topic, the decryption key and validity date details will be entered automatically on receipt of a UBX RXM-SPARTNKEY message (which may take a few seconds after connecting). Alternatively, they can be uploaded from the service provider's JSON file, or entered manually as follows: -1. Enter the current and next decryption keys in hexadecimal format e.g. `0102030405060708090a0b0c0d0e0f10`. The keys are normally 16 bytes long, or 32 hexadecimal characters. -1. Enter the supplied Valid From dates in `YYYYMMDD` format. **NB:** These are *Valid From* dates rather than *Expiry* dates. If the location service provides Expiry dates, subtract 4 weeks from these to get the Valid From dates. -1. Select 'DGPS Timeout', 'Upload keys', 'Configure receiver' and 'Disable NMEA' options as required. -1. Select the SPARTN correction source - either `IP` or `L-Band`. -1. To reduce traffic volumes, you can choose to disable NMEA messages from the receiver. A minimal set of UBX NAV messages will be enabled in their place. -1. Click ![send button](https://github.com/semuconsulting/PyGPSClient/blob/master/src/pygpsclient/resources/iconmonstr-arrow-12-24.png?raw=true) to upload the configuration. The utility will send the relevant configuration command(s) to the receiver and poll for an acknowledgement. - - -Below is a illustrative SPARTN DGPS data log, showing: -* Incoming SPARTN correction messages (SPARTN-1X-HPAC* high-precision atmosphere corrections; SPARTN-OCB* orbit / clock / bias corrections; SPARTN-1X-GAD geographic area definitions). -* Outgoing UBX RXM-COR confirmation messages from receiver showing that the SPARTN data has been received and decrypted OK. - -![spartn console screenshot](https://github.com/semuconsulting/PyGPSClient/blob/master/images/spartn_consolelog.png?raw=true) \ No newline at end of file diff --git a/docs/pygpsclient.rst b/docs/pygpsclient.rst index 9da9c236..650f0e38 100644 --- a/docs/pygpsclient.rst +++ b/docs/pygpsclient.rst @@ -372,46 +372,6 @@ pygpsclient.socketconfig\_ntrip\_frame module :undoc-members: :show-inheritance: -pygpsclient.spartn\_dialog module ---------------------------------- - -.. automodule:: pygpsclient.spartn_dialog - :members: - :undoc-members: - :show-inheritance: - -pygpsclient.spartn\_gnss\_frame module --------------------------------------- - -.. automodule:: pygpsclient.spartn_gnss_frame - :members: - :undoc-members: - :show-inheritance: - -pygpsclient.spartn\_json\_config module ---------------------------------------- - -.. automodule:: pygpsclient.spartn_json_config - :members: - :undoc-members: - :show-inheritance: - -pygpsclient.spartn\_lband\_frame module ---------------------------------------- - -.. automodule:: pygpsclient.spartn_lband_frame - :members: - :undoc-members: - :show-inheritance: - -pygpsclient.spartn\_mqtt\_frame module --------------------------------------- - -.. automodule:: pygpsclient.spartn_mqtt_frame - :members: - :undoc-members: - :show-inheritance: - pygpsclient.spectrum\_frame module ---------------------------------- diff --git a/pygpsclient.json b/pygpsclient.json index 28a61a1c..db4d9cc8 100644 --- a/pygpsclient.json +++ b/pygpsclient.json @@ -107,37 +107,6 @@ "ntripclientreflon_f": -115.81513, "ntripclientrefalt_f": 15.0, "ntripclientrefsep_f": 0.0, - "mqttclientserver_s": "", - "mqttclientport_n": 8883, - "mqttclientid_s": "<<=== YOUR MQTT CLIENT ID ===>", - "mqttclientregion_s": "eu", - "mqttclientmode_n": 0, - "mqttclienttopicip_b": 1, - "mqttclienttopicmga_b": 1, - "mqttclienttopickey_b": 1, - "mqttclienttlscrt_s": "<=== FULLY QUALIFIED PATH TO MQTT CRT FILE ===>", - "mqttclienttlskey_s": "<=== FULLY QUALIFIED PATH TO MQTT KEY FILE ===>", - "lbandclientbpsrate_n": 9600, - "lbandclientdatabits_n": 8, - "lbandclientstopbits_f": 1.0, - "lbandclientparity_s": "N", - "lbandclientrtscts_b": 0, - "lbandclientxonxoff_b": 0, - "lbandclienttimeout_f": 0.1, - "lbandclientmsgmode_n": 0, - "lband_enabled_b": 1, - "spartnport_s": "<== YOUR USER_DEFINED D9S SERIAL PORT ==>", - "lbandclientfreq_n": 1556290000, - "lbandclientschwin_n": 2200, - "lbandclientsid_n": 21845, - "lbandclientdrat_n": 2400, - "lbandclientusesid_b": 0, - "lbandclientdescrm_b": 1, - "lbandclientprescrm_b": 0, - "lbandclientdescrminit_n": 26969, - "lbandclientunqword_s": "16238547128276412563", - "lbandclientoutport_s": "Passthrough", - "lbandclientdebug_b": 0, "spartndecode_b": 0, "spartnkey_s": "abcd1234abcd1234abcd1234abcd1234", "spartnbasedate_n": -1, diff --git a/src/pygpsclient/_version.py b/src/pygpsclient/_version.py index 07f89dde..24f75b4c 100644 --- a/src/pygpsclient/_version.py +++ b/src/pygpsclient/_version.py @@ -8,4 +8,4 @@ :license: BSD 3-Clause """ -__version__ = "1.7.1" +__version__ = "1.7.2" diff --git a/src/pygpsclient/app.py b/src/pygpsclient/app.py index c36816ca..96c31b1b 100644 --- a/src/pygpsclient/app.py +++ b/src/pygpsclient/app.py @@ -255,7 +255,8 @@ def __init__(self, **kwargs): # display any deferred messages if isinstance(self._deferredmsg, tuple): - self.set_status_label(self._deferredmsg) + msg, col = self._deferredmsg + self.set_status_label(msg, col) self._deferredmsg = None # check for more recent version (if enabled) @@ -583,6 +584,10 @@ def refresh_widgets(self): frm.update_frame() self.update() + # update database if enabled (must be done in main App thread) + if self.configuration.get("database_b"): + self.sqlite_handler.load_data() + if self.conn_status != DISCONNECTED or self.rtk_conn_status != DISCONNECTED: update_interval = int(self.configuration.get("guiupdateinterval_f") * 1000) self.refresh_widget_timer = self.after( diff --git a/src/pygpsclient/configuration.py b/src/pygpsclient/configuration.py index 9e0f1278..3c8d8fc6 100644 --- a/src/pygpsclient/configuration.py +++ b/src/pygpsclient/configuration.py @@ -31,7 +31,6 @@ CUSTOM, DDD, DEFAULT_PASSWORD, - DEFAULT_REGION, DEFAULT_USER, ERRCOL, FORMAT_BINARY, @@ -39,19 +38,13 @@ GUI_UPDATE_INTERVAL, MAXLOGSIZE, MIN_GUI_UPDATE_INTERVAL, - MQTTIPMODE, OKCOL, - PASSTHRU, RCVR_CONNECTION, SOCKCLIENT_HOST, SOCKCLIENT_PORT, SOCKSERVER_HOST, SOCKSERVER_NTRIP_PORT, SOCKSERVER_PORT, - SPARTN_BASEDATE_CURRENT, - SPARTN_DEFAULT_KEY, - SPARTN_OUTPORT, - SPARTN_PPSERVER_URL, TRACK, UBLOX_ZEDF9, UMM, @@ -60,7 +53,6 @@ from pygpsclient.init_presets import INIT_PRESETS from pygpsclient.mapquest_handler import MAP_UPDATE_INTERVAL from pygpsclient.serverconfig_dialog import BASE_SVIN -from pygpsclient.spartn_lband_frame import D9S_PP_EU as D9S_PP from pygpsclient.strings import ( LOADCONFIGBAD, LOADCONFIGNK, @@ -208,45 +200,9 @@ def __init__(self, app): "ntripclientreflon_f": 0.0, "ntripclientrefalt_f": 0.0, "ntripclientrefsep_f": 0.0, - # SPARTN MQTT (IP) client settings from pygnssutils.GNSSMQTTClient - "mqttclientserver_s": SPARTN_PPSERVER_URL, - "mqttclientport_n": SPARTN_OUTPORT, - "mqttclientid_s": "<<=== YOUR MQTT CLIENT ID ===>", - "mqttclientregion_s": DEFAULT_REGION, - "mqttclientmode_n": MQTTIPMODE, - "mqttclienttopicip_b": 1, - "mqttclienttopicmga_b": 1, - "mqttclienttopickey_b": 1, - "mqttclienttopicfreq_b": 0, - "mqttclienttlscrt_s": "<=== FULLY QUALIFIED PATH TO MQTT CRT FILE ===>", - "mqttclienttlskey_s": "<=== FULLY QUALIFIED PATH TO MQTT KEY FILE ===>", - # SPARTN L-Band client settings from SpartnLbandDialog if open - "lband_enabled_b": 1, # SPARTN L-Band and MQTT services discontinued - "spartnport_s": "", "spartndecode_b": 0, - "spartnkey_s": SPARTN_DEFAULT_KEY, - "spartnbasedate_n": SPARTN_BASEDATE_CURRENT, - "lbandclientserialport_s": "/dev/ttyACM1", - "lbandclientbpsrate_n": 9600, - "lbandclientdatabits_n": 8, - "lbandclientstopbits_f": 1.0, - "lbandclientparity_s": PARITY_NONE, - "lbandclientrtscts_b": 0, - "lbandclientxonxoff_b": 0, - "lbandclienttimeout_f": 0.1, - "lbandclientmsgmode_n": GET, - "lbandclientinactivity_timeout_n": 0, - "lbandclientfreq_n": D9S_PP["freq"], - "lbandclientschwin_n": D9S_PP["schwin"], - "lbandclientsid_n": D9S_PP["sid"], - "lbandclientdrat_n": 2400, - "lbandclientusesid_b": 0, - "lbandclientdescrm_b": 1, - "lbandclientprescrm_b": 0, - "lbandclientdescrminit_n": D9S_PP["descrminit"], - "lbandclientunqword_s": D9S_PP["unqword"], - "lbandclientoutport_s": PASSTHRU, - "lbandclientdebug_b": 0, + "spartnkey_s": "abcd1234abcd1234abcd1234abcd1234", + "spartnbasedate_n": -1, "relposnedsettings_d": { "source_s": "NAV-RELPOSNED", }, diff --git a/src/pygpsclient/dialog_state.py b/src/pygpsclient/dialog_state.py index 2a872d14..791e70ae 100644 --- a/src/pygpsclient/dialog_state.py +++ b/src/pygpsclient/dialog_state.py @@ -25,7 +25,6 @@ from pygpsclient.rinex_dialog import RINEXDialog from pygpsclient.serverconfig_dialog import ServerConfigDialog from pygpsclient.settings_dialog import SettingsDialog -from pygpsclient.spartn_dialog import SPARTNConfigDialog from pygpsclient.strings import ( DLG, DLGTABOUT, @@ -37,7 +36,6 @@ DLGTRINEX, DLGTSERVER, DLGTSETTINGS, - DLGTSPARTN, DLGTTTY, DLGTUBX, DLGTUBXLEGACY, @@ -88,11 +86,6 @@ def __init__(self): DLG: None, RESIZE: True, }, - DLGTSPARTN: { - CLASS: SPARTNConfigDialog, - DLG: None, - RESIZE: False, - }, DLGTGPX: { CLASS: GPXViewerDialog, DLG: None, diff --git a/src/pygpsclient/menu_bar.py b/src/pygpsclient/menu_bar.py index c8b992ca..8a39959f 100644 --- a/src/pygpsclient/menu_bar.py +++ b/src/pygpsclient/menu_bar.py @@ -14,7 +14,6 @@ from tkinter import Menu -from pygpsclient.strings import DLGTSPARTN # service discontinued by u-blox from pygpsclient.strings import ( DLGTABOUT, DLGTGPX, @@ -44,7 +43,6 @@ DLGTNMEA, DLGTNTRIP, DLGTSERVER, - DLGTSPARTN, # service discontinued by u-blox DLGTGPX, DLGTIMPORTMAP, DLGTTTY, diff --git a/src/pygpsclient/spartn_dialog.py b/src/pygpsclient/spartn_dialog.py deleted file mode 100644 index 21006474..00000000 --- a/src/pygpsclient/spartn_dialog.py +++ /dev/null @@ -1,248 +0,0 @@ -""" -spartn_dialog.py - -SPARTN client configuration dialog. - -Initial settings are from the saved configuration. -Once started, the persisted state for the SPARTN client is held in -the threaded SPARTN handler itself, NOT in this frame. - -The dialog may be closed while the SPARTN client is running. - -Created on 26 Jan 2023 - -:author: semuadmin (Steve Smith) -:copyright: 2020 semuadmin -:license: BSD 3-Clause -""" - -from tkinter import NSEW -from types import NoneType - -from pyubx2 import UBXMessage - -from pygpsclient.globals import ( - CONNECTED_SPARTNIP, - CONNECTED_SPARTNLB, - SPARTN_GNSS, - SPARTN_LBAND, - SPARTN_MQTT, -) -from pygpsclient.spartn_gnss_frame import SPARTNGNSSDialog -from pygpsclient.spartn_lband_frame import SpartnLbandDialog -from pygpsclient.spartn_mqtt_frame import SPARTNMQTTDialog -from pygpsclient.strings import DLGTSPARTN -from pygpsclient.toplevel_dialog import ToplevelDialog - -RXMMSG = "RXM-SPARTN-KEY" -CFGSET = "CFG-VALGET/SET" -CFGPOLL = "CFG-VALGET" - - -class SPARTNConfigDialog(ToplevelDialog): - """, - SPARTNConfigDialog class. - """ - - def __init__(self, app, *args, **kwargs): # pylint: disable=unused-argument - """ - Constructor. - - :param Tk app: reference to main tkinter application - :param args: optional args to pass to parent class (not currently used) - :param kwargs: optional kwargs to pass to parent class (not currently used) - """ - - self.__app = app # Reference to main application class - - super().__init__(app, DLGTSPARTN) - self._pending_confs = {} - self._lband_enabled = self.__app.configuration.get("lband_enabled_b") - - self._body() - self._do_layout() - self._reset() - self._attach_events() - self._finalise() - - def _body(self): - """ - Set up frame and widgets. - """ - # pylint: disable=unnecessary-lambda - - self.frm_corrip = SPARTNMQTTDialog( - self.__app, - self, - borderwidth=2, - relief="groove", - ) - if self._lband_enabled: - self.frm_corrlband = SpartnLbandDialog( - self.__app, - self, - borderwidth=2, - relief="groove", - ) - self.frm_gnss = SPARTNGNSSDialog( - self.__app, self, borderwidth=2, relief="groove" - ) - - def _do_layout(self): - """ - Position widgets in frame. - """ - - self.frm_corrip.grid( - column=0, - row=0, - ipadx=5, - ipady=5, - sticky=NSEW, - ) - col = 1 - if self._lband_enabled: - self.frm_corrlband.grid( - column=col, - row=0, - ipadx=5, - ipady=5, - sticky=NSEW, - ) - col += 1 - self.frm_gnss.grid( - column=col, - row=0, - ipadx=5, - ipady=5, - sticky=NSEW, - ) - - def _reset(self): - """ - Reset configuration widgets. - """ - - self.status_label = "" - - def _attach_events(self): - """ - Bind events to window. - """ - - # self.bind("", self._on_resize) - - def set_pending(self, msgid: int, spartnfrm: int): - """ - Set pending confirmation flag for UBX configuration frame to - signify that it's waiting for a confirmation message. - - :param int msgid: UBX message identity - :param int spartnfrm: integer representing SPARTN configuration frame - """ - - self._pending_confs[msgid] = spartnfrm - - def update_pending(self, msg: UBXMessage): - """ - Update pending confirmation status. - :param UBXMessage msg: UBX config message - """ - - if not hasattr(msg, "identity"): - return - - # update ebno & fecBits values - if msg.identity == "RXM-PMP": - self.frm_corrlband.update_status(msg) - return - - spartnfrm = self._pending_confs.get(msg.identity, None) - - if spartnfrm is not None: - if spartnfrm == SPARTN_GNSS: - self.frm_gnss.update_status(msg) - elif spartnfrm == SPARTN_LBAND: - self.frm_corrlband.update_status(msg) - elif spartnfrm == SPARTN_MQTT: - self.frm_corrip.update_status(msg) - - # reset all confirmation flags for this frame - for msgid in (msg.identity, "ACK-ACK", "ACK-NAK"): - if self._pending_confs.get(msgid, None) == spartnfrm: - self._pending_confs.pop(msgid) - - def set_controls(self, status: bool, msgt: tuple | NoneType = None): - """ - Set controls in IP or L-Band clients. - - :param bool status: connected to SPARTN server yes/no - :param tuple | Nonetype msgt: tuple of (message, color) or None - """ - - if status == CONNECTED_SPARTNIP: - self.frm_corrip.set_controls(status) - elif status == CONNECTED_SPARTNLB: - self.frm_corrlband.set_controls(status) - if msgt is not None: - self.set_status_label(msgt) - - def disconnect_ip(self, msg: str = ""): - """ - Disconnect from IP (MQTT) client. - - :param str msg: optional disconnection message - """ - - self.frm_corrip.on_disconnect(msg) - - def disconnect_lband(self, msg: str = ""): - """ - Disconnect from L-Band client. - - :param str msg: optional disconnection message - """ - - self.frm_corrlband.on_disconnect(msg) - - @property - def server(self) -> str: - """ - Getter for server. - - :return: server - :rtype: str - """ - - return self.frm_corrip.server - - @server.setter - def server(self, server: str): - """ - Setter for server. - - :param str clientid: Client ID - """ - - self.frm_corrip.server = server - - @property - def clientid(self) -> str: - """ - Getter for Client ID from IP configuration dialog. - - :return: client ID - :rtype: str - """ - - return self.frm_corrip.clientid - - @clientid.setter - def clientid(self, clientid: str): - """ - Setter for Client ID. - - :param str clientid: Client ID - """ - - self.frm_corrip.clientid = clientid diff --git a/src/pygpsclient/spartn_gnss_frame.py b/src/pygpsclient/spartn_gnss_frame.py deleted file mode 100644 index 46dfde15..00000000 --- a/src/pygpsclient/spartn_gnss_frame.py +++ /dev/null @@ -1,542 +0,0 @@ -""" -spartn_gnss_frame.py - -SPARTN GNSS Receiver configuration dialog. - -This handles the configuration of the GNSS receiver to receive and decrypt -incoming SPARTN RTK messages. - -FYI CFG_NAVSPG_CONSTR_DGNSSTO governs how quickly rcvr drops out of RTK mode -after loss of correction signal - default is 60 seconds but this may need -to be increased if the correction source is intermittent. - -Created on 26 Jan 2023 - -:author: semuadmin (Steve Smith) -:copyright: 2020 semuadmin -:license: BSD 3-Clause -""" - -from datetime import datetime, timedelta, timezone -from tkinter import ( - EW, - NORMAL, - Button, - Checkbutton, - Entry, - Frame, - IntVar, - Label, - Radiobutton, - Spinbox, - StringVar, - Tk, - W, - ttk, -) - -from PIL import Image, ImageTk -from pyspartn import date2timetag -from pyubx2 import POLL, SET, SET_LAYER_RAM, TXN_NONE, U1, U2, U4, UBXMessage, val2bytes - -from pygpsclient.globals import ( - CLICK_CURSOR, - CONNECTED, - ERRCOL, - ICON_BLANK, - ICON_CONFIRMED, - ICON_DISCONN, - ICON_LOAD, - ICON_PENDING, - ICON_SEND, - ICON_WARNING, - OKCOL, - READONLY, - RXMMSG, - SPARTN_GNSS, - SPARTN_KEYLEN, - SPARTN_SOURCE_IP, - SPARTN_SOURCE_LB, - TRACEMODE_WRITE, - VALDMY, - VALHEX, - VALLEN, -) -from pygpsclient.helpers import validate # pylint: disable=unused-import -from pygpsclient.helpers import ( - date2wnotow, - parse_rxmspartnkey, -) -from pygpsclient.spartn_json_config import SpartnJsonConfig -from pygpsclient.strings import ( - CONFIGBAD, - CONFIGOK, - CONFIGRXM, - DLGJSONERR, - DLGJSONOK, - LBLDISNMEA, - LBLJSONLOAD, - LBLSPARTNGN, - LBLSPTNCURR, - LBLSPTNDAT, - LBLSPTNFP, - LBLSPTNKEY, - LBLSPTNNEXT, - LBLSPTNUPLOAD, - NOTCONN, - NULLSEND, -) - -U2MAX = 2e16 - 1 -U8MAX = 2e64 - 1 -BAUDRATE = 38400 -RESERVED0 = b"\x00\x00" -RESERVED1 = b"\x00" -CFGSET = "CFG-VALGET/SET" -CFGPOLL = "CFG-VALGET" -INPORTS = ("I2C", "UART1", "UART2", "USB", "SPI") -RTK_FLOAT = 2 -RTK_FIXED = 3 - - -class SPARTNGNSSDialog(Frame): - """, - SPARTNConfigDialog class. - """ - - def __init__(self, app: Tk, parent: Frame, *args, **kwargs): - """ - Constructor. - - :param Tk app: reference to main tkinter application - :param Frame container: reference to container frame - :param args: optional args to pass to parent class (not currently used) - :param kwargs: optional kwargs to pass to parent class (not currently used) - """ - - self.__app = app # Reference to main application class - self.__container = parent # container frame - - super().__init__(parent.container, *args, **kwargs) - - self._img_blank = ImageTk.PhotoImage(Image.open(ICON_BLANK)) - self._img_pending = ImageTk.PhotoImage(Image.open(ICON_PENDING)) - self._img_confirmed = ImageTk.PhotoImage(Image.open(ICON_CONFIRMED)) - self._img_warn = ImageTk.PhotoImage(Image.open(ICON_WARNING)) - self._img_send = ImageTk.PhotoImage(Image.open(ICON_SEND)) - self._img_disconn = ImageTk.PhotoImage(Image.open(ICON_DISCONN)) - self._img_load = ImageTk.PhotoImage(Image.open(ICON_LOAD)) - self._status = StringVar() - self._spartn_source = IntVar() - self._status_cfgmsg = StringVar() - self._spartn_key1 = StringVar() - self._spartn_valdate1 = StringVar() - self._spartn_key2 = StringVar() - self._spartn_valdate2 = StringVar() - self._upload_keys = IntVar() - self._send_f9p_config = IntVar() - self._dgnssto = IntVar() - self._disable_nmea = IntVar() - self._ports = INPORTS - - self._body() - self._do_layout() - self._reset() - self._attach_events() - - def _body(self): - """ - Set up frame and widgets. - """ - # pylint: disable=unnecessary-lambda - - # GNSS Receiver (F9P) configuration widgets - self._lbl_gnss = Label(self, text=LBLSPARTNGN) - self._lbl_curr = Label(self, text=LBLSPTNCURR) - self._lbl_key1 = Label(self, text=LBLSPTNKEY) - self._ent_key1 = Entry( - self, - textvariable=self._spartn_key1, - state=NORMAL, - relief="sunken", - width=32, - ) - self._lbl_valdate1 = Label(self, text=LBLSPTNDAT) - self._ent_valdate1 = Entry( - self, - textvariable=self._spartn_valdate1, - state=NORMAL, - relief="sunken", - width=8, - ) - self._lbl_next = Label(self, text=LBLSPTNNEXT) - self._lbl_key2 = Label(self, text=LBLSPTNKEY) - self._ent_key2 = Entry( - self, - textvariable=self._spartn_key2, - state=NORMAL, - relief="sunken", - width=32, - ) - self._lbl_valdate2 = Label(self, text=LBLSPTNDAT) - self._ent_valdate2 = Entry( - self, - textvariable=self._spartn_valdate2, - state=NORMAL, - relief="sunken", - width=8, - ) - self._rad_source0 = Radiobutton( - self, text="IP", variable=self._spartn_source, value=SPARTN_SOURCE_IP - ) - self._rad_source1 = Radiobutton( - self, - text="L-Band", - variable=self._spartn_source, - value=SPARTN_SOURCE_LB, - ) - self._lbl_dgnssto = Label( - self, - text="DGPS Timeout", - ) - self._spn_dgnssto = Spinbox( - self, - values=(60, 120, 240), - width=4, - state=READONLY, - wrap=True, - textvariable=self._dgnssto, - ) - self._lbl_loadjson = Label( - self, - text=LBLJSONLOAD, - ) - self._btn_loadjson = Button( - self, - width=45, - image=self._img_load, - command=lambda: self._on_load_json(), - cursor=CLICK_CURSOR, - ) - self._chk_upload_keys = Checkbutton( - self, - text=LBLSPTNUPLOAD, - variable=self._upload_keys, - ) - self._chk_send_config = Checkbutton( - self, - text=LBLSPTNFP, - variable=self._send_f9p_config, - ) - self._chk_disable_nmea = Checkbutton( - self, text=LBLDISNMEA, variable=self._disable_nmea - ) - self._lbl_send_command = Label(self, image=self._img_blank) - self._btn_send_command = Button( - self, - image=self._img_send, - width=45, - fg=OKCOL, - command=self._on_send_gnss_config, - font=self.__app.font_md, - cursor=CLICK_CURSOR, - ) - - def _do_layout(self): - """ - Position widgets in frame. - """ - - self._lbl_gnss.grid(column=0, row=0, columnspan=4, padx=3, pady=2, sticky=W) - self._lbl_loadjson.grid(column=0, row=1, padx=3, pady=2, sticky=W) - self._btn_loadjson.grid(column=1, row=1, columnspan=3, padx=3, pady=2, sticky=W) - ttk.Separator(self).grid( - column=0, row=2, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._lbl_curr.grid(column=0, row=3, columnspan=4, padx=3, pady=2, sticky=W) - self._lbl_key1.grid(column=0, row=4, padx=3, pady=2, sticky=W) - self._ent_key1.grid(column=1, row=4, columnspan=3, padx=3, pady=2, sticky=W) - self._lbl_valdate1.grid(column=0, row=5, padx=3, pady=2, sticky=W) - self._ent_valdate1.grid(column=1, row=5, columnspan=3, padx=3, pady=2, sticky=W) - ttk.Separator(self).grid( - column=0, row=6, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._lbl_next.grid(column=0, row=7, columnspan=3, padx=3, pady=2, sticky=W) - self._lbl_key2.grid(column=0, row=8, padx=3, pady=2, sticky=W) - self._ent_key2.grid(column=1, row=8, columnspan=3, padx=2, pady=2, sticky=W) - self._lbl_valdate2.grid(column=0, row=9, padx=3, pady=2, sticky=W) - self._ent_valdate2.grid(column=1, row=9, columnspan=3, padx=3, pady=2, sticky=W) - ttk.Separator(self).grid( - column=0, row=10, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._rad_source0.grid(column=0, row=11, padx=3, pady=2, sticky=W) - self._rad_source1.grid(column=1, row=11, columnspan=3, padx=3, pady=2, sticky=W) - self._lbl_dgnssto.grid(column=0, row=12, columnspan=3, padx=3, pady=2, sticky=W) - self._spn_dgnssto.grid(column=1, row=12, columnspan=3, padx=3, pady=2, sticky=W) - self._chk_upload_keys.grid(column=0, row=13, padx=3, pady=2, sticky=W) - self._chk_send_config.grid(column=1, row=13, padx=3, pady=2, sticky=W) - self._chk_disable_nmea.grid( - column=2, row=13, columnspan=2, padx=3, pady=2, sticky=W - ) - ttk.Separator(self).grid( - column=0, row=14, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._btn_send_command.grid(column=2, row=15, padx=3, pady=2, sticky=W) - self._lbl_send_command.grid(column=3, row=15, padx=3, pady=2, sticky=W) - - def _reset(self): - """ - Reset configuration widgets. - """ - - curr = datetime.now() - self._set_date(self._spartn_valdate1, curr) - self._set_date(self._spartn_valdate2, curr + timedelta(weeks=4)) - self._spartn_source.set(SPARTN_SOURCE_IP) - self._upload_keys.set(1) - self._send_f9p_config.set(1) - self._disable_nmea.set(0) - - self._poll_config() - - def _attach_events(self): - """ - Set up event listeners. - """ - - self._spartn_key1.trace_add(TRACEMODE_WRITE, self._on_update_config) - - def _on_update_config(self, var, index, mode): # pylint: disable=unused-argument - """ - Update in-memory configuration if setting is changed. - """ - - self.update() - self.__app.configuration.set("spartnkey_s", self._spartn_key1.get()) - self.__app.configuration.set( - "spartnbasedate_n", date2timetag(datetime.now(timezone.utc)) - ) - - def _valid_gnss_settings(self) -> bool: - """ - Validate GNSS receiver settings. - - :return: valid True/False - :rtype: bool - """ - - valid = True - # key must be valid hexadecimal and 16 bytes (32 chars) in length - valid = ( - valid - & self._ent_key1.validate(VALHEX) - & self._ent_key1.validate(VALLEN, SPARTN_KEYLEN * 2, SPARTN_KEYLEN * 2) - ) - # from date must be valid date in format YYYYMMDD - valid = valid & self._ent_valdate1.validate(VALDMY) - valid = ( - valid - & self._ent_key2.validate(VALHEX) - & self._ent_key2.validate(VALLEN, SPARTN_KEYLEN * 2, SPARTN_KEYLEN * 2) - ) - valid = valid & self._ent_valdate2.validate(VALDMY) - - if not valid: - self.__container.status_label = ("ERROR - invalid settings", ERRCOL) - - return valid - - def _format_cfggnss(self) -> UBXMessage: - """ - Format UBX CFG-VALSET message to configure GNSS receiver. - """ - - cfgdata = [] - cfgdata.append(("CFG_NAVHPG_DGNSSMODE", RTK_FIXED)) - # select SPARTN source (L-band or IP) - cfgdata.append(("CFG_SPARTN_USE_SOURCE", self._spartn_source.get())) - # set DGNSS Timeout (default = 60, higher can be better with L-Band) - cfgdata.append(("CFG_NAVSPG_CONSTR_DGNSSTO", self._dgnssto.get())) - # enable SPARTN and UBX on selected ports - for port in self._ports: - cfgdata.append((f"CFG_{port}INPROT_UBX", 1)) - cfgdata.append((f"CFG_{port}INPROT_RTCM3X", 1)) - cfgdata.append((f"CFG_{port}INPROT_SPARTN", 1)) - cfgdata.append((f"CFG_MSGOUT_UBX_RXM_COR_{port}", 1)) - - # if NMEA disabled, remove it from out ports and enable - # a couple of UBX NAV messages instead - if self._disable_nmea.get(): - cfgdata.append((f"CFG_{port}OUTPROT_NMEA", 0)) - cfgdata.append((f"CFG_MSGOUT_UBX_NAV_PVT_{port}", 1)) - cfgdata.append((f"CFG_MSGOUT_UBX_NAV_DOP_{port}", 1)) - cfgdata.append((f"CFG_MSGOUT_UBX_NAV_SAT_{port}", 4)) - else: - cfgdata.append((f"CFG_{port}OUTPROT_NMEA", 1)) - # cfgdata.append((f"CFG_MSGOUT_UBX_NAV_PVT_{port}", 0)) - # cfgdata.append((f"CFG_MSGOUT_UBX_NAV_DOP_{port}", 0)) - # cfgdata.append((f"CFG_MSGOUT_UBX_NAV_SAT_{port}", 0)) - - msg = UBXMessage.config_set(SET_LAYER_RAM, TXN_NONE, cfgdata) - return msg - - def _format_rxmspartn(self) -> UBXMessage: - """ - Format UBX RXM-SPARTN-KEY message. - """ - - version = val2bytes(1, U1) - numKeys = val2bytes(2, U1) - key1 = bytes.fromhex(self._spartn_key1.get()) - keylen1 = val2bytes(len(key1), U1) - wno1, tow1 = date2wnotow(self._get_date(self._spartn_valdate1)) - wno1b = val2bytes(wno1, U2) - tow1b = val2bytes(tow1, U4) - key2 = bytes.fromhex(self._spartn_key2.get()) - keylen2 = val2bytes(len(key2), U1) - wno2, tow2 = date2wnotow(self._get_date(self._spartn_valdate2)) - wno2b = val2bytes(wno2, U2) - tow2b = val2bytes(tow2, U4) - - payload = ( - version - + numKeys - + RESERVED0 - + RESERVED1 - + keylen1 - + wno1b - + tow1b - + RESERVED1 - + keylen2 - + wno2b - + tow2b - + key1 - + key2 - ) - msg = UBXMessage("RXM", RXMMSG, SET, payload=payload) - return msg - - def _on_send_gnss_config(self): - """ - Send config to GNSS receiver (e.g. F9P). - """ - - if not self._valid_gnss_settings(): - return - - if self.__app.conn_status != CONNECTED: - self.__container.status_label = (NOTCONN, ERRCOL) - - if self._send_f9p_config.get(): - msgc = "config" - msg = self._format_cfggnss() - self._send_gnss_command(msg) - else: - msgc = "" - if self._upload_keys.get(): - msgk = "keys" - msg = self._format_rxmspartn() - self._send_gnss_command(msg) - # poll RXM-SPARTN-KEY to check keys have loaded - msg = UBXMessage("RXM", RXMMSG, POLL) - self._send_gnss_command(msg) - else: - msgk = "" - if msgk == "" and msgc == "": - self.__container.status_label = (NULLSEND, ERRCOL) - return - if msgk != "" and msgc != "": - msga = " and " - else: - msga = "" - self.__container.status_label = f"{(msgk + msga + msgc).capitalize()} sent" - self._lbl_send_command["image"] = self._img_pending - for msgid in ("RXM-SPARTNKEY", "CFG-VALGET", "ACK-ACK", "ACK-NAK"): - self.__container.set_pending(msgid, SPARTN_GNSS) - - def _send_gnss_command(self, msg: UBXMessage): - """ - Send command to GNSS receiver. - """ - - self.__app.send_to_device(msg.serialize()) - - def _set_date(self, val: StringVar, dat: datetime): - """ - Set date string in entry field. - """ - - val.set(dat.strftime("%Y%m%d")) - - def _get_date(self, val: StringVar) -> datetime: - """ - Get datetime from entry field. - """ - - dat = val.get() - return datetime(int(dat[0:4]), int(dat[4:6]), int(dat[6:8])) - - def _poll_config(self): - """ - Poll receiver for current RXM-SPARTN-KEY configuration. - """ - - msg = UBXMessage("RXM", RXMMSG, POLL) - self.__app.send_to_device(msg.serialize()) - self.__container.set_pending(RXMMSG, SPARTN_GNSS) - - def update_status(self, msg: UBXMessage): - """ - Update pending confirmation status. - :param UBXMessage msg: UBX config message - """ - - if msg.identity == RXMMSG: - self._lbl_send_command["image"] = self._img_confirmed - if msg.numKeys == 2: - keydata = parse_rxmspartnkey(msg) # key1, date1, key2, date2 - self._spartn_key1.set(keydata[0][0]) - self._spartn_valdate1.set(keydata[0][1].strftime("%Y%m%d")) - self._spartn_key2.set(keydata[1][0]) - self._spartn_valdate2.set(keydata[1][1].strftime("%Y%m%d")) - # save latest key in configuration settings - self.__app.configuration.set("spartnkey_s", self._spartn_key1.get()) - col = OKCOL - else: - col = ERRCOL - self.__container.status_label = (CONFIGRXM.format(RXMMSG, msg.numKeys), col) - elif msg.identity == "ACK-ACK": - self._lbl_send_command["image"] = self._img_confirmed - self.__container.status_label = (CONFIGOK.format(CFGSET), OKCOL) - elif msg.identity == "ACK-NAK": - self._lbl_send_command["image"] = self._img_warn - self.__container.status_label = (CONFIGBAD.format(CFGSET), ERRCOL) - self.update_idletasks() - - def _on_load_json(self): - """ - Load SPARTN decryption keys from JSON file. - """ - # pylint: disable=unused-variable - - jsonfile = self.__app.file_handler.open_file( - self, "spartnjson", (("json files", "*.json"), ("all files", "*.*")) - ) - if jsonfile is None: - return - - try: - spc = SpartnJsonConfig(jsonfile) - # strip scheme and port from server name - scheme, host, port = spc.server.split(":") - self.__container.server = host.replace("//", "") - self.__container.clientid = spc.clientid - key, start, _ = spc.current_key - self._spartn_key1.set(key) - self._spartn_valdate1.set(start.strftime("%Y%m%d")) - key, start, _ = spc.next_key - self._spartn_key2.set(key) - self._spartn_valdate2.set(start.strftime("%Y%m%d")) - self.__container.status_label = (DLGJSONOK.format(jsonfile), OKCOL) - except Exception as err: # pylint: disable=broad-exception-caught - self.__container.status_label = (DLGJSONERR.format(err), ERRCOL) diff --git a/src/pygpsclient/spartn_json_config.py b/src/pygpsclient/spartn_json_config.py deleted file mode 100644 index 52014972..00000000 --- a/src/pygpsclient/spartn_json_config.py +++ /dev/null @@ -1,161 +0,0 @@ -""" -spartn_json_config.py - -Utility class to load SPARTN decryption configuration from -JSON file provided by ThingStream PointPerfect location service. -Requires paid subscription. - -JSON file normally named "device-{Client ID}-ucenter-config.json" - -Created on 12 Feb 2023 - -:author: semuadmin (Steve Smith) -:copyright: 2020 semuadmin -:license: BSD 3-Clause -""" - -import json -from datetime import datetime, timedelta - - -class SpartnJsonConfig: - """ - SpartnJsonConfig class. - """ - - def __init__(self, filename): - """ - Constructor. - - :param str filename: Fully qualified path to JSON file - """ - - if filename in (None, ""): - raise AttributeError("Filename must be provided") - - self._loadconfig(filename) - - def _loadconfig(self, filename: str): - """ - Import config from JSON file. - - :param str filename: filename - """ - - with open(filename, "r", encoding="utf-8") as jsonfile: - jsondict = json.load(jsonfile) - connection = jsondict["MQTT"]["Connectivity"] - subscriptions = jsondict["MQTT"]["Subscriptions"] - dynamic = jsondict["MQTT"]["dynamickeys"] - - self._clientid = connection["ClientID"] - self._server = connection["ServerURI"] - self._key = connection["ClientCredentials"]["Key"] - self._cert = connection["ClientCredentials"]["Cert"] - self._rootca = connection["ClientCredentials"]["RootCA"] - - self._topics = {} - self._topics["Key"] = tuple(subscriptions["Key"]["KeyTopics"]) - self._topics["AssistNow"] = tuple( - subscriptions["AssistNow"]["AssistNowTopics"] - ) - self._topics["Data"] = tuple(subscriptions["Data"]["DataTopics"]) - - current_ts = dynamic["current"]["start"] - current_dur = dynamic["current"]["duration"] - next_ts = dynamic["next"]["start"] - next_dur = dynamic["next"]["duration"] - self._curr_key = dynamic["current"]["value"] - self._curr_start = datetime.fromtimestamp(current_ts / 1000) - self._curr_end = self._curr_start + timedelta(milliseconds=current_dur) - self._next_key = dynamic["next"]["value"] - self._next_start = datetime.fromtimestamp(next_ts / 1000) - self._next_end = self._next_start + timedelta(milliseconds=next_dur) - - @property - def clientid(self) -> str: - """ - Getter for ClientID. - - :return: clientID - :rtype: str - """ - - return self._clientid - - @property - def server(self) -> str: - """ - Getter for Server URI. - - :return: Server UIR - :rtype: str - """ - - return self._server - - @property - def topics(self) -> list: - """ - Getter for topics. - - :return: list of topics - :rtype: list - """ - - return self._topics - - @property - def key(self) -> str: - """ - Getter for Key. - - :return: Key - :rtype: str - """ - - return self._key - - @property - def cert(self) -> str: - """ - Getter for Certificate. - - :return: Certificate - :rtype: str - """ - - return self._cert - - @property - def rootca(self) -> str: - """ - Getter for RootCA. - - :return: RootCA - :rtype: str - """ - - return self._rootca - - @property - def current_key(self) -> tuple: - """ - Getter for current key. - - :return: tuple of (key, start date, end date) - :rtype: tuple - """ - - return (self._curr_key, self._curr_start, self._curr_end) - - @property - def next_key(self) -> tuple: - """ - Getter for next key. - - :return: tuple of (key, start date, end date) - :rtype: tuple - """ - - return (self._next_key, self._next_start, self._next_end) diff --git a/src/pygpsclient/spartn_lband_frame.py b/src/pygpsclient/spartn_lband_frame.py deleted file mode 100644 index 5da6832f..00000000 --- a/src/pygpsclient/spartn_lband_frame.py +++ /dev/null @@ -1,730 +0,0 @@ -""" -spartn_lband_frame.py - -SPARTN L-Band Correction receiver configuration dialog. - -This handles the configuration of the L-Band receiver -(e.g. u-blox D9S) for the selected region. - -Supply initial settings via `config` keyword argument. - -Created on 26 Jan 2023 - -:author: semuadmin (Steve Smith) -:copyright: 2020 semuadmin -:license: BSD 3-Clause -""" - -from tkinter import ( - DISABLED, - EW, - NORMAL, - NSEW, - Button, - Checkbutton, - Entry, - Frame, - IntVar, - Label, - Spinbox, - StringVar, - TclError, - Tk, - W, - ttk, -) - -from PIL import Image, ImageTk -from pyubx2 import POLL_LAYER_RAM, SET, SET_LAYER_RAM, TXN_NONE, UBXMessage - -from pygpsclient.globals import ( - BPSRATES, - CLICK_CURSOR, - CONNECTED, - CONNECTED_SPARTNIP, - CONNECTED_SPARTNLB, - DISCONNECTED, - ERRCOL, - ICON_BLANK, - ICON_CONFIRMED, - ICON_DISCONN, - ICON_EXIT, - ICON_PENDING, - ICON_SEND, - ICON_SERIAL, - ICON_SOCKET, - ICON_WARNING, - KNOWNGPS, - MSGMODES, - NOPORTS, - OKCOL, - PASSTHRU, - READONLY, - SPARTN_EOF_EVENT, - SPARTN_ERR_EVENT, - SPARTN_EVENT, - SPARTN_LBAND, - TIMEOUTS, - TRACEMODE_WRITE, - VALINT, -) -from pygpsclient.helpers import validate # pylint: disable=unused-import -from pygpsclient.serialconfig_lband_frame import SerialConfigLbandFrame -from pygpsclient.strings import CONFIGBAD, CONFIGOK, DLGSPARTNWARN, LBLSPARTNLB - -U2MAX = 2e16 - 1 -U8MAX = 2e64 - 1 -BAUDRATE = 38400 -SPARTN_KEYLEN = 16 -RESERVED0 = b"\x00\x00" -RESERVED1 = b"\x00" -CFGSET = "CFG-VALGET/SET" -CFGPOLL = "CFG-VALGET" -INPORTS = ("I2C", "UART1", "UART2", "USB", "SPI") -PMP_DATARATES = { - "B600": 600, - "B1200": 1200, - "B2400": 2400, - "B4800": 4800, -} -D9S_FACTORY = { - "name": "D9S Factory Default", - "freq": 1539812500, - "schwin": 2200, - "usesid": 1, - "sid": 50821, - "drat": PMP_DATARATES["B2400"], - "descrm": 1, - "prescrm": 1, - "descrminit": 23560, - "unqword": "16238547128276412563", -} -D9S_PP_US = { - "name": "D9S PointPerfect US", - "freq": 1556290000, - "schwin": 2200, - "usesid": 0, - "sid": 21845, - "drat": PMP_DATARATES["B2400"], - "descrm": 1, - "prescrm": 0, - "descrminit": 26969, - "unqword": "16238547128276412563", -} -D9S_PP_EU = { - "name": "D9S PointPerfect EU", - "freq": 1545260000, - "schwin": 2200, - "usesid": 0, - "sid": 21845, - "drat": PMP_DATARATES["B2400"], - "descrm": 1, - "prescrm": 0, - "descrminit": 26969, - "unqword": "16238547128276412563", -} -# D9S default configuration -D9S_CONFIG = D9S_PP_US # D9S_FACTORY - - -class SpartnLbandDialog(Frame): - """, - SPARTNConfigDialog class. - """ - - def __init__(self, app: Tk, parent: Frame, *args, **kwargs): - """ - Constructor. - - :param Tk app: reference to main tkinter application - :param Frame parent: reference to parent frame - :param args: optional args to pass to parent class (not currently used) - :param kwargs: optional kwargs to pass to parent class (not currently used) - """ - - self.__app = app # Reference to main application class - self.__container = parent # container frame - - super().__init__(parent.container, *args, **kwargs) - - self._img_blank = ImageTk.PhotoImage(Image.open(ICON_BLANK)) - self._img_pending = ImageTk.PhotoImage(Image.open(ICON_PENDING)) - self._img_confirmed = ImageTk.PhotoImage(Image.open(ICON_CONFIRMED)) - self._img_warn = ImageTk.PhotoImage(Image.open(ICON_WARNING)) - self._img_exit = ImageTk.PhotoImage(Image.open(ICON_EXIT)) - self._img_send = ImageTk.PhotoImage(Image.open(ICON_SEND)) - self._img_serial = ImageTk.PhotoImage(Image.open(ICON_SERIAL)) - self._img_socket = ImageTk.PhotoImage(Image.open(ICON_SOCKET)) - self._img_disconn = ImageTk.PhotoImage(Image.open(ICON_DISCONN)) - self._status = StringVar() - self._spartn_freq = StringVar() - self._spartn_schwin = StringVar() - self._spartn_usesid = IntVar() - self._spartn_sid = StringVar() - self._spartn_drat = StringVar() - self._spartn_descrm = IntVar() - self._spartn_prescrm = IntVar() - self._spartn_descrminit = StringVar() - self._spartn_unqword = StringVar() - self._spartn_outport = StringVar() - self._enabledbg = IntVar() - self._saveconfig = IntVar() - self._ports = INPORTS - self._settings = {} - self._connected = DISCONNECTED - - self._body() - self._do_layout() - self._reset() - self._attach_events() - - def _body(self): - """ - Set up frame and widgets. - """ - # pylint: disable=unnecessary-lambda - self._lbl_corrlband = Label(self, text=LBLSPARTNLB) - - # Correction receiver serial port configuration panel - self._frm_spartn_serial = SerialConfigLbandFrame( - self.__app, - self, - recognised=KNOWNGPS, - timeouts=TIMEOUTS, - bpsrates=BPSRATES, - msgmodes=list(MSGMODES.keys()), - ) - self._lbl_freq = Label(self, text="L-Band Frequency (Hz)") - self._ent_freq = Entry( - self, - textvariable=self._spartn_freq, - state=NORMAL, - relief="sunken", - width=10, - ) - self._chk_enabledbg = Checkbutton( - self, - text="Enable Debug?", - variable=self._enabledbg, - ) - self._lbl_schwin = Label(self, text="Search window") - self._ent_schwin = Entry( - self, - textvariable=self._spartn_schwin, - state=NORMAL, - relief="sunken", - width=10, - ) - self._chk_saveconfig = Checkbutton( - self, - text="Save Config?", - variable=self._saveconfig, - ) - self._chk_usesid = Checkbutton( - self, - text="Use Service ID?", - variable=self._spartn_usesid, - ) - self._lbl_sid = Label(self, text="Service ID") - self._ent_sid = Entry( - self, - textvariable=self._spartn_sid, - state=NORMAL, - relief="sunken", - width=10, - ) - self._lbl_drat = Label(self, text="Data Rate") - self._spn_drat = Spinbox( - self, - values=list(PMP_DATARATES.values()), - textvariable=self._spartn_drat, - state=READONLY, - width=5, - wrap=True, - ) - self._chk_descrm = Checkbutton( - self, - text="Use Descrambler?", - variable=self._spartn_descrm, - ) - self._chk_prescram = Checkbutton( - self, - text="Use Prescrambling?", - variable=self._spartn_prescrm, - ) - self._lbl_descraminit = Label(self, text="Descrambler Init") - self._ent_descraminit = Entry( - self, - textvariable=self._spartn_descrminit, - state=NORMAL, - relief="sunken", - width=10, - ) - self._lbl_unqword = Label(self, text="Unique Word") - self._ent_unqword = Entry( - self, - textvariable=self._spartn_unqword, - state=NORMAL, - relief="sunken", - width=20, - ) - self._lbl_outport = Label(self, text="Output Port") - self._spn_outport = Spinbox( - self, - values=INPORTS + (PASSTHRU,), - textvariable=self._spartn_outport, - state=READONLY, - width=10, - wrap=True, - ) - self._lbl_ebno = Label(self, text="") - self._lbl_fec = Label(self, text="") - self._btn_connect = Button( - self, - width=45, - image=self._img_serial, - command=lambda: self.on_connect(), - cursor=CLICK_CURSOR, - ) - self._btn_disconnect = Button( - self, - width=45, - image=self._img_disconn, - command=lambda: self.on_disconnect(), - state=DISABLED, - cursor=CLICK_CURSOR, - ) - self._lbl_send = Label(self, image=self._img_blank) - self._btn_send = Button( - self, - image=self._img_send, - width=45, - command=self._on_send_config, - cursor=CLICK_CURSOR, - ) - - def _do_layout(self): - """ - Position widgets in frame. - """ - - self._lbl_corrlband.grid( - column=0, row=0, columnspan=4, padx=3, pady=2, sticky=W - ) - self._frm_spartn_serial.grid( - column=0, row=1, columnspan=4, padx=3, pady=2, sticky=NSEW - ) - ttk.Separator(self).grid( - column=0, row=2, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._lbl_freq.grid(column=0, row=3, sticky=W) - self._ent_freq.grid(column=1, row=3, sticky=W) - self._chk_enabledbg.grid(column=2, row=3, columnspan=2, sticky=W) - self._lbl_schwin.grid(column=0, row=4, sticky=W) - self._ent_schwin.grid(column=1, row=4, sticky=W) - self._chk_saveconfig.grid(column=2, row=4, sticky=W) - self._chk_usesid.grid(column=0, row=5, sticky=W) - self._chk_descrm.grid(column=1, row=5, sticky=W) - self._chk_prescram.grid(column=2, row=5, columnspan=2, sticky=W) - self._lbl_sid.grid(column=0, row=6, sticky=W) - self._ent_sid.grid(column=1, row=6, columnspan=3, sticky=W) - self._lbl_drat.grid(column=0, row=7, sticky=W) - self._spn_drat.grid(column=1, row=7, columnspan=3, sticky=W) - self._lbl_descraminit.grid(column=0, row=8, sticky=W) - self._ent_descraminit.grid(column=1, row=8, columnspan=3, sticky=W) - self._lbl_unqword.grid(column=0, row=9, sticky=W) - self._ent_unqword.grid(column=1, row=9, columnspan=3, sticky=W) - self._lbl_outport.grid(column=0, row=10, sticky=W) - self._spn_outport.grid(column=1, row=10, columnspan=3, sticky=W) - self._lbl_ebno.grid(column=0, row=11, sticky=W) - self._lbl_fec.grid(column=1, row=11, sticky=W) - ttk.Separator(self).grid( - column=0, row=12, columnspan=4, padx=2, pady=3, sticky=EW - ) - self._btn_connect.grid(column=0, row=13, padx=3, pady=2, sticky=W) - self._btn_disconnect.grid(column=1, row=13, padx=3, pady=2, sticky=W) - self._btn_send.grid(column=2, row=13, padx=3, pady=2, sticky=W) - self._lbl_send.grid(column=3, row=13, padx=3, pady=2, sticky=W) - - def _attach_events(self): - """ - Bind events to frame. - """ - - self.__app.bind(SPARTN_ERR_EVENT, self.on_error) - for setting in ( - self._enabledbg, - self._spartn_freq, - self._spartn_schwin, - self._spartn_usesid, - self._spartn_sid, - self._spartn_drat, - self._spartn_descrm, - self._spartn_prescrm, - self._spartn_descrminit, - self._spartn_unqword, - self._spartn_outport, - ): - setting.trace_add(TRACEMODE_WRITE, self._on_update_config) - - def _reset(self): - """ - Reset configuration widgets. - """ - - cfg = self.__app.configuration - self._saveconfig.set(0) - self._enabledbg.set(cfg.get("lbandclientdebug_b")) - self._spartn_drat.set(cfg.get("lbandclientdrat_n")) - self._spartn_freq.set(cfg.get("lbandclientfreq_n")) - self._spartn_schwin.set(cfg.get("lbandclientschwin_n")) - self._spartn_usesid.set(cfg.get("lbandclientusesid_b")) - self._spartn_sid.set(cfg.get("lbandclientsid_n")) - self._spartn_descrm.set(cfg.get("lbandclientdescrm_b")) - self._spartn_prescrm.set(cfg.get("lbandclientprescrm_b")) - self._spartn_descrminit.set(cfg.get("lbandclientdescrminit_n")) - self._spartn_unqword.set(cfg.get("lbandclientunqword_s")) - self._spartn_outport.set(cfg.get("lbandclientoutport_s")) - self.__container.status_label = "" - if self.__app.rtk_conn_status == CONNECTED_SPARTNLB: - self.set_controls(CONNECTED_SPARTNLB) - else: - self.set_controls(DISCONNECTED) - - def _on_update_config(self, var, index, mode): # pylint: disable=unused-argument - """ - Update in-memory configuration if setting is changed. - """ - - self.update() - cfg = self.__app.configuration - try: - cfg.set("lbandclientdebug_b", int(self._enabledbg.get())) - cfg.set("lbandclientdrat_n", int(self._spartn_drat.get())) - cfg.set("lbandclientfreq_n", int(self._spartn_freq.get())) - cfg.set("lbandclientschwin_n", int(self._spartn_schwin.get())) - cfg.set("lbandclientusesid_b", int(self._spartn_usesid.get())) - cfg.set("lbandclientsid_n", int(self._spartn_sid.get())) - cfg.set("lbandclientdescrm_b", int(self._spartn_descrm.get())) - cfg.set("lbandclientprescrm_b", int(self._spartn_prescrm.get())) - cfg.set("lbandclientdescrminit_n", int(self._spartn_descrminit.get())) - cfg.set("lbandclientunqword_s", self._spartn_unqword.get()) - cfg.set("lbandclientoutport_s", self._spartn_outport.get()) - except (ValueError, TclError): - pass - - def set_controls(self, status: int): - """ - Enable or disable widgets depending on connection status. - - :param int status: connection status (0 = disconnected, 1 = connected) - """ - - self._frm_spartn_serial.status_label = status - stat = DISABLED if status == CONNECTED_SPARTNLB else NORMAL - for wdg in (self._btn_connect,): - wdg["state"] = stat - stat = NORMAL if status == CONNECTED_SPARTNLB else DISABLED - for wdg in ( - self._ent_freq, - self._ent_schwin, - self._ent_sid, - self._ent_unqword, - self._ent_descraminit, - self._chk_enabledbg, - self._chk_saveconfig, - self._chk_descrm, - self._chk_prescram, - self._chk_usesid, - self._btn_disconnect, - self._btn_send, - ): - wdg["state"] = stat - stat = READONLY if status == CONNECTED_SPARTNLB else DISABLED - for wdg in ( - self._spn_drat, - self._spn_outport, - ): - wdg["state"] = stat - - def _valid_settings(self) -> bool: - """ - Validate settings. - - :return: valid True/False - :rtype: bool - """ - - valid = True - valid = valid & self._ent_freq.validate(VALINT, 1e9, 2e9) # L-band 1GHz - 2GHz - valid = valid & self._ent_schwin.validate(VALINT, 0, U2MAX) # U2 - valid = valid & self._ent_sid.validate(VALINT, 0, U2MAX) # U2 - valid = valid & self._ent_descraminit.validate(VALINT, 0, U2MAX) # U2 - valid = valid & self._ent_unqword.validate(VALINT, 0, U8MAX) # U8 - - if not valid: - self.__container.status_label = ("ERROR - invalid settings", ERRCOL) - - return valid - - def on_connect(self): - """ - Connect to Correction receiver. - """ - - frm = self._frm_spartn_serial - if self.__app.rtk_conn_status == CONNECTED_SPARTNIP: - self.__container.status_label = ( - DLGSPARTNWARN.format("IP", "L-Band"), - ERRCOL, - ) - return - - if frm.status == NOPORTS: - return - - conndict = { - "protocol": self.__app.protocol_mask, - "read_event": SPARTN_EVENT, - "eof_event": SPARTN_EOF_EVENT, - "error_event": SPARTN_ERR_EVENT, - "inqueue": self.__app.spartn_inqueue, - "outqueue": self.__app.spartn_outqueue, - "socket_inqueue": self.__app.socket_inqueue, - "conntype": CONNECTED, - "msgmode": self.__app.configuration.get("msgmode_n"), - "serial_settings": self._frm_spartn_serial, - } - - # start serial stream thread - self.__app.rtk_conn_status = CONNECTED_SPARTNLB - self.__app.spartn_stream_handler.start(self.__container, conndict) - self.__container.status_label = ( - f"Connected to {frm.port}:{frm.port_desc} @ {frm.bpsrate}" - ) - self.set_controls(CONNECTED_SPARTNLB) - - self._poll_config() - - def on_disconnect(self, msg: str = ""): - """ - Disconnect from L-Band Correction receiver. - - :param str msg: optional disconnection message - """ - - msg += "Disconnected" - if self.__app.rtk_conn_status == CONNECTED_SPARTNLB: - if self.__app.spartn_stream_handler is not None: - self.__app.spartn_stream_handler.stop() - self.__app.rtk_conn_status = DISCONNECTED - self.__container.status_label = (msg, ERRCOL) - self.set_controls(DISCONNECTED) - - def on_error(self, event): # pylint: disable=unused-argument - """ - EVENT TRIGGERED - Action on <> event - disconnect. - - :param event event: read event - """ - - self.on_disconnect() - - def _format_cfgpoll(self) -> UBXMessage: - """ - Format UBX CFG-VALGET message to poll Correction receiver - configuration. - """ - - outport = self._spartn_outport.get() - if outport == PASSTHRU: - outport = "USB" - - cfgdata = [ - "CFG_PMP_CENTER_FREQUENCY", - "CFG_PMP_SEARCH_WINDOW", - "CFG_PMP_USE_SERVICE_ID", - "CFG_PMP_SERVICE_ID", - "CFG_PMP_DATA_RATE", - "CFG_PMP_USE_DESCRAMBLER", - "CFG_PMP_DESCRAMBLER_INIT", - "CFG_PMP_USE_PRESCRAMBLING", - "CFG_PMP_UNIQUE_WORD", - ] - for port in INPORTS: - cfgdata.append(f"CFG_MSGOUT_UBX_RXM_PMP_{port}") - cfgdata.append(f"CFG_{port}OUTPROT_UBX") - if port in ("UART1", "UART2"): - cfgdata.append(f"CFG_{port}_BAUDRATE") - - msg = UBXMessage.config_poll(POLL_LAYER_RAM, 0, cfgdata) - return msg - - def _format_cfgcorr(self) -> UBXMessage: - """ - Format UBX CFG-VALSET message to configure Correction receiver. - Sets RCM-PMP and selection output port configuration. - """ - - outport = self._spartn_outport.get() - if outport == PASSTHRU: - outport = "USB" - - cfgdata = [ - ("CFG_PMP_CENTER_FREQUENCY", int(self._spartn_freq.get())), - ("CFG_PMP_SEARCH_WINDOW", int(self._spartn_schwin.get())), - ("CFG_PMP_USE_SERVICE_ID", int(self._spartn_usesid.get())), - ("CFG_PMP_SERVICE_ID", int(self._spartn_sid.get())), - ("CFG_PMP_DATA_RATE", int(self._spartn_drat.get())), - ("CFG_PMP_USE_DESCRAMBLER", int(self._spartn_descrm.get())), - ("CFG_PMP_DESCRAMBLER_INIT", int(self._spartn_descrminit.get())), - ("CFG_PMP_USE_PRESCRAMBLING", int(self._spartn_prescrm.get())), - ("CFG_PMP_UNIQUE_WORD", int(self._spartn_unqword.get())), - ] - for port in INPORTS: - cfgdata.append((f"CFG_MSGOUT_UBX_RXM_PMP_{port}", 1)) - cfgdata.append((f"CFG_{port}OUTPROT_UBX", 1)) - if self._enabledbg.get(): - cfgdata.append((f"CFG_INFMSG_UBX_{port}", b"\x1f")) - else: - cfgdata.append((f"CFG_INFMSG_UBX_{port}", b"\x07")) - if outport in ("UART1", "UART2"): - cfgdata.append( - f"CFG_{outport}_BAUDRATE", int(self._frm_spartn_serial.bpsrate) - ) - - msg = UBXMessage.config_set(SET_LAYER_RAM, TXN_NONE, cfgdata) - return msg - - def _format_cfgcfg(self) -> UBXMessage: - """ - Format UBX CFG-CFG command to persist configuration - to correction receiver's BBR / Flash / EEPROM memory. - """ - - msg = UBXMessage( - "CFG", - "CFG-CFG", - SET, - # clearMask=b"\x01\x00\x00\x00", - saveMask=b"\x1f\x1f\x00\x00", - # loadMask=b"\x01\x00\x00\x00", - devBBR=1, - devFlash=1, - devEEPROM=1, - ) - return msg - - def _on_send_config(self): - """ - Send config to L-Band Correction receiver (e.g. D9S). - """ - - if not self._valid_settings(): - return - - msg = self._format_cfgcorr() - self._send_command(msg) - self.__container.status_label = f"{CFGSET} command sent" - self._lbl_send["image"] = self._img_pending - - # save config to persistent memory - if self._saveconfig.get(): - msg = self._format_cfgcfg() - self._send_command(msg) - - self._poll_config() - - def _poll_config(self): - """ - Poll receiver for current configuration. - """ - - msg = self._format_cfgpoll() - self._send_command(msg) - - for msgid in (CFGPOLL, "ACK-ACK", "ACK-NAK"): - self.__container.set_pending(msgid, SPARTN_LBAND) - - def _send_command(self, msg: UBXMessage): - """ - Send command to L-Band Correction receiver. - """ - - self.__app.spartn_outqueue.put(msg.serialize()) - - def update_status(self, msg: UBXMessage): - """ - Update pending confirmation status. - :param UBXMessage msg: UBX config message - """ - - if msg.identity == CFGPOLL: - if hasattr(msg, "CFG_PMP_CENTER_FREQUENCY"): - self._spartn_freq.set(msg.CFG_PMP_CENTER_FREQUENCY) - if hasattr(msg, "CFG_PMP_SEARCH_WINDOW"): - self._spartn_schwin.set(msg.CFG_PMP_SEARCH_WINDOW) - if hasattr(msg, "CFG_PMP_USE_SERVICE_ID"): - self._spartn_usesid.set(msg.CFG_PMP_USE_SERVICE_ID) - if hasattr(msg, "CFG_PMP_SERVICE_ID"): - self._spartn_sid.set(msg.CFG_PMP_SERVICE_ID) - if hasattr(msg, "CFG_PMP_DATA_RATE"): - self._spartn_drat.set(msg.CFG_PMP_DATA_RATE) - if hasattr(msg, "CFG_PMP_USE_DESCRAMBLER"): - self._spartn_descrm.set(msg.CFG_PMP_USE_DESCRAMBLER) - if hasattr(msg, "CFG_PMP_DESCRAMBLER_INIT"): - self._spartn_descrminit.set(msg.CFG_PMP_DESCRAMBLER_INIT) - if hasattr(msg, "CFG_PMP_USE_PRESCRAMBLING"): - self._spartn_prescrm.set(msg.CFG_PMP_USE_PRESCRAMBLING) - if hasattr(msg, "CFG_PMP_UNIQUE_WORD"): - self._spartn_unqword.set(msg.CFG_PMP_UNIQUE_WORD) - self.__container.status_label = (f"{CFGPOLL} received", OKCOL) - self._lbl_send["image"] = self._img_confirmed - elif msg.identity == "ACK-ACK": - self._lbl_send["image"] = self._img_confirmed - self.__container.status_label = (CONFIGOK.format(CFGSET), OKCOL) - elif msg.identity == "ACK-NAK": - self._lbl_send["image"] = self._img_warn - self.__container.status_label = (CONFIGBAD.format(CFGSET), ERRCOL) - elif msg.identity == "RXM-PMP": - self._lbl_ebno["text"] = f"Eb/N0: {msg.ebno} dB" - self._lbl_fec["text"] = f"FEC Bits: {msg.fecBits}" - - self.update_idletasks() - - @property - def settings(self) -> dict: - """ - Getter for settings. - - :return: dictionary of settings - :rtype: dict - """ - - try: - settings = { - "bpsrate": self._frm_spartn_serial.bpsrate, - "databits": self._frm_spartn_serial.databits, - "stopbits": self._frm_spartn_serial.stopbits, - "parity": self._frm_spartn_serial.parity, - "rtscts": self._frm_spartn_serial.rtscts, - "xonxoff": self._frm_spartn_serial.xonxoff, - "timeout": self._frm_spartn_serial.timeout, - "msgmode": self._frm_spartn_serial.msgmode, - "userport": self._frm_spartn_serial.userport, - "freq": int(self._spartn_freq.get()), - "schwin": int(self._spartn_schwin.get()), - "usesid": int(self._spartn_usesid.get()), - "sid": int(self._spartn_sid.get()), - "drat": int(self._spartn_drat.get()), - "descrm": int(self._spartn_descrm.get()), - "prescrm": int(self._spartn_prescrm.get()), - "descrminit": int(self._spartn_descrminit.get()), - "unqword": self._spartn_unqword.get(), - "outport": self._spartn_outport.get(), - "debug": int(self._enabledbg.get()), - } - return settings - except (TypeError, ValueError): - return {} diff --git a/src/pygpsclient/spartn_mqtt_frame.py b/src/pygpsclient/spartn_mqtt_frame.py deleted file mode 100644 index b06a6174..00000000 --- a/src/pygpsclient/spartn_mqtt_frame.py +++ /dev/null @@ -1,618 +0,0 @@ -""" -spartn_mqtt_frame.py - -SPARTN configuration dialog - -This is the pop-up dialog containing the various -SPARTN configuration functions. - -NB: The initial configuration for the SPARTN MQTT client -(pygnssutils.GNSSMQTTClient) is set in app.update_SPARTN_handler(). -Once started, the persisted state for the SPARTN MQTT client is -held in the threaded SPARTN handler itself, NOT in this frame. - -Created on 26 Jan 2023 - -:author: semuadmin (Steve Smith) -:copyright: 2020 semuadmin -:license: BSD 3-Clause -""" - -from os import path -from pathlib import Path -from tkinter import ( - DISABLED, - EW, - NORMAL, - Button, - Checkbutton, - E, - Entry, - Frame, - IntVar, - Label, - Radiobutton, - Spinbox, - StringVar, - TclError, - Tk, - W, - ttk, -) - -from PIL import Image, ImageTk -from pyspartn import TIMEBASE - -try: - from pyspartn import HASCRYPTO -except ImportError: - HASCRYPTO = 1 -from pyubx2 import UBXMessage - -from pygpsclient.globals import ( - CLICK_CURSOR, - CONNECTED_SPARTNIP, - CONNECTED_SPARTNLB, - DISCONNECTED, - ERRCOL, - ICON_BLANK, - ICON_CONFIRMED, - ICON_DISCONN, - ICON_EXIT, - ICON_LOAD, - ICON_PENDING, - ICON_SEND, - ICON_SERIAL, - ICON_SOCKET, - ICON_WARNING, - MQTTIPMODE, - MQTTLBANDMODE, - OKCOL, - READONLY, - RXMMSG, - SPARTN_BASEDATE_CURRENT, - SPARTN_BASEDATE_DATASTREAM, - SPARTN_GNSS, - SPARTN_OUTPORT, - SPARTN_PPREGIONS, - TRACEMODE_WRITE, - VALINT, - VALLEN, -) -from pygpsclient.helpers import MAXPORT, validate # pylint: disable=unused-import -from pygpsclient.strings import DLGSPARTNWARN, LBLSPARTNIP, MQTTCONN - - -class SPARTNMQTTDialog(Frame): - """, - SPARTNConfigDialog class. - """ - - def __init__(self, app: Tk, parent: Frame, *args, **kwargs): - """ - Constructor. - - :param Tk app: reference to main tkinter application - :param Frame parent: reference to parent frame - :param args: optional args to pass to parent class (not currently used) - :param kwargs: optional kwargs to pass to parent class (not currently used) - """ - - self.__app = app # Reference to main application class - self.__container = parent # container frame - - super().__init__(parent.container, *args, **kwargs) - - self._img_blank = ImageTk.PhotoImage(Image.open(ICON_BLANK)) - self._img_pending = ImageTk.PhotoImage(Image.open(ICON_PENDING)) - self._img_confirmed = ImageTk.PhotoImage(Image.open(ICON_CONFIRMED)) - self._img_warn = ImageTk.PhotoImage(Image.open(ICON_WARNING)) - self._img_exit = ImageTk.PhotoImage(Image.open(ICON_EXIT)) - self._img_send = ImageTk.PhotoImage(Image.open(ICON_SEND)) - self._img_serial = ImageTk.PhotoImage(Image.open(ICON_SERIAL)) - self._img_socket = ImageTk.PhotoImage(Image.open(ICON_SOCKET)) - self._img_disconn = ImageTk.PhotoImage(Image.open(ICON_DISCONN)) - self._img_load = ImageTk.PhotoImage(Image.open(ICON_LOAD)) - self._stream_handler = None - self._status = StringVar() - self._mqtt_server = StringVar() - self._mqtt_port = IntVar() - self._mqtt_port.set(SPARTN_OUTPORT) - self._mqtt_region = StringVar() - self._mqtt_source = IntVar() - self._mqtt_clientid = StringVar() - self._mqtt_crt = StringVar() - self._mqtt_pem = StringVar() - self._mqtt_iptopic = IntVar() - self._mqtt_mgatopic = IntVar() - self._mqtt_keytopic = IntVar() - self._mqtt_freqtopic = IntVar() - self._spartndecode = IntVar() - self._settings = {} - # spartn_inqueue is read by app and passed to GNSS receiver - self._output = self.__app.spartn_inqueue - self._connected = DISCONNECTED - - self._body() - self._do_layout() - self._reset() - self._attach_events() - - def _body(self): - """ - Set up frame and widgets. - """ - # pylint: disable=unnecessary-lambda - - self._lbl_corrip = Label(self, text=LBLSPARTNIP) - self._lbl_mqttserver = Label(self, text="Server") - self._ent_mqttserver = Entry( - self, - textvariable=self._mqtt_server, - state=NORMAL, - relief="sunken", - width=25, - ) - self._lbl_mqttport = Label(self, text="Port") - self._ent_mqttport = Entry( - self, - textvariable=self._mqtt_port, - state=NORMAL, - relief="sunken", - width=8, - ) - self._lbl_mqttregion = Label(self, text="Region") - self._spn_mqttregion = Spinbox( - self, - values=SPARTN_PPREGIONS, - textvariable=self._mqtt_region, - state=READONLY, - width=4, - wrap=True, - ) - self._rad_ip = Radiobutton( - self, text="IP", variable=self._mqtt_source, value=MQTTIPMODE - ) - self._rad_lb = Radiobutton( - self, text="L-Band", variable=self._mqtt_source, value=MQTTLBANDMODE - ) - self._lbl_mqttclientid = Label(self, text="Client ID") - self._ent_mqttclientid = Entry( - self, - textvariable=self._mqtt_clientid, - state=NORMAL, - relief="sunken", - width=35, - ) - self._lbl_topics = Label(self, text="Topics") - self._chk_mqtt_iptopic = Checkbutton( - self, - text="IP", - variable=self._mqtt_iptopic, - ) - self._chk_mqtt_mgatopic = Checkbutton( - self, - text="Assist", - variable=self._mqtt_mgatopic, - ) - self._chk_mqtt_keytopic = Checkbutton( - self, - text="Key", - variable=self._mqtt_keytopic, - ) - self._chk_mqtt_freqtopic = Checkbutton( - self, - text="Freq", - variable=self._mqtt_freqtopic, - ) - self._btn_opencrt = Button( - self, - width=45, - image=self._img_load, - command=lambda: self._get_spartncerts("crt"), - cursor=CLICK_CURSOR, - ) - self._lbl_mqttcrt = Label(self, text="CRT File") - self._ent_mqttcrt = Entry( - self, - textvariable=self._mqtt_crt, - state=NORMAL, - relief="sunken", - width=32, - ) - self._btn_openpem = Button( - self, - width=45, - image=self._img_load, - command=lambda: self._get_spartncerts("pem"), - cursor=CLICK_CURSOR, - ) - self._lbl_mqttpem = Label(self, text="PEM File") - self._ent_mqttpem = Entry( - self, - textvariable=self._mqtt_pem, - state=NORMAL, - relief="sunken", - width=32, - ) - self._lbl_spartndecode = Label(self, text="Decode SPARTN in console") - self._chk_spartndecode = Checkbutton( - self, - text="", - variable=self._spartndecode, - ) - self._btn_connect = Button( - self, - width=45, - image=self._img_socket, - command=lambda: self.on_connect(), - cursor=CLICK_CURSOR, - ) - self._btn_disconnect = Button( - self, - width=45, - image=self._img_disconn, - command=lambda: self.on_disconnect(), - state=DISABLED, - cursor=CLICK_CURSOR, - ) - - def _do_layout(self): - """ - Position widgets in frame. - """ - - self._lbl_corrip.grid(column=0, row=0, columnspan=6, padx=3, pady=2, sticky=W) - self._lbl_mqttserver.grid(column=0, row=1, padx=3, pady=2, sticky=W) - self._ent_mqttserver.grid( - column=1, row=1, columnspan=5, padx=3, pady=2, sticky=W - ) - self._lbl_mqttport.grid(column=0, row=2, padx=3, pady=2, sticky=W) - self._ent_mqttport.grid(column=1, row=2, padx=3, pady=2, sticky=W) - self._lbl_mqttregion.grid(column=0, row=3, padx=3, pady=2, sticky=W) - self._spn_mqttregion.grid(column=1, row=3, padx=3, pady=2, sticky=W) - self._rad_ip.grid(column=2, row=3, padx=3, pady=2, sticky=W) - self._rad_lb.grid(column=3, row=3, padx=3, pady=2, sticky=W) - self._lbl_mqttclientid.grid(column=0, row=4, padx=3, pady=2, sticky=W) - self._ent_mqttclientid.grid( - column=1, row=4, columnspan=5, padx=3, pady=2, sticky=W - ) - self._lbl_topics.grid(column=0, row=5, padx=4, pady=2, sticky=W) - self._chk_mqtt_iptopic.grid(column=1, row=5, padx=3, pady=2, sticky=W) - self._chk_mqtt_mgatopic.grid(column=2, row=5, padx=3, pady=2, sticky=W) - self._chk_mqtt_keytopic.grid(column=3, row=5, padx=3, pady=2, sticky=W) - self._chk_mqtt_freqtopic.grid(column=4, row=5, padx=3, pady=2, sticky=W) - self._lbl_mqttcrt.grid(column=0, row=6, padx=3, pady=2, sticky=W) - self._ent_mqttcrt.grid(column=1, row=6, padx=3, columnspan=3, pady=2, sticky=EW) - self._btn_opencrt.grid(column=4, row=6, padx=3, pady=2, sticky=E) - self._lbl_mqttpem.grid(column=0, row=7, padx=3, pady=2, sticky=W) - self._ent_mqttpem.grid(column=1, row=7, columnspan=3, padx=3, pady=2, sticky=EW) - self._btn_openpem.grid(column=4, row=7, padx=3, pady=2, sticky=E) - self._lbl_spartndecode.grid(column=0, row=8, columnspan=2, sticky=W) - self._chk_spartndecode.grid(column=2, row=8, sticky=W) - ttk.Separator(self).grid( - column=0, row=9, columnspan=6, padx=2, pady=3, sticky=EW - ) - self._btn_connect.grid(column=0, row=10, padx=3, pady=2, sticky=W) - self._btn_disconnect.grid(column=2, row=10, padx=3, pady=2, sticky=W) - - def _attach_events(self): - """ - Set up event listeners. - """ - - self._mqtt_source.trace_add(TRACEMODE_WRITE, self._on_source) - for setting in ( - self._mqtt_server, - self._mqtt_port, - self._mqtt_region, - self._mqtt_clientid, - self._mqtt_crt, - self._mqtt_pem, - self._mqtt_iptopic, - self._mqtt_mgatopic, - self._mqtt_keytopic, - self._mqtt_freqtopic, - self._spartndecode, - ): - setting.trace_add(TRACEMODE_WRITE, self._on_update_config) - - def _reset(self): - """ - Reset configuration widgets. - """ - - self._get_settings() - self._reset_keypaths(self._mqtt_clientid.get()) - self.__container.status_label = "" - if self.__app.rtk_conn_status == CONNECTED_SPARTNIP: - self.set_controls(CONNECTED_SPARTNIP) - else: - self.set_controls(DISCONNECTED) - - if not HASCRYPTO: - self._lbl_spartndecode["state"] = DISABLED - self._chk_spartndecode["state"] = DISABLED - - def _reset_keypaths(self, clientid): - """ - Reset key and cert file paths. - - :param str clientid: Client ID - """ - - spartn_crt = path.join(Path.home(), f"device-{clientid}-pp-cert.crt") - spartn_pem = path.join(Path.home(), f"device-{clientid}-pp-key.pem") - self._mqtt_crt.set(spartn_crt) - self._mqtt_pem.set(spartn_pem) - - def _on_source(self, var, index, mode): # pylint: disable=unused-argument - """ - Action when source is changed. - """ - - self.update() - cfg = self.__app.configuration - if self._mqtt_source.get() == MQTTLBANDMODE: - self._chk_mqtt_iptopic["state"] = DISABLED - self._chk_mqtt_freqtopic["state"] = NORMAL - self._mqtt_iptopic.set(0) - self._mqtt_freqtopic.set(1) - else: # MQTTIPMODE - self._chk_mqtt_iptopic["state"] = NORMAL - self._chk_mqtt_freqtopic["state"] = DISABLED - self._mqtt_iptopic.set(1) - self._mqtt_freqtopic.set(0) - cfg.set("mqttclientmode_n", self._mqtt_source.get()) - - def _on_update_config(self, var, index, mode): # pylint: disable=unused-argument - """ - Update in-memory configuration if setting is changed. - """ - - self.update() - cfg = self.__app.configuration - try: - cfg.set("mqttclientserver_s", self._mqtt_server.get()) - cfg.set("mqttclientport_n", int(self._mqtt_port.get())) - cfg.set("mqttclientregion_s", self._mqtt_region.get()) - cfg.set("mqttclientid_s", self._mqtt_clientid.get()) - cfg.set("mqttclienttlscrt_s", self._mqtt_crt.get()) - cfg.set("mqttclienttlskey_s", self._mqtt_pem.get()) - cfg.set("mqttclienttopicip_b", int(self._mqtt_iptopic.get())) - cfg.set("mqttclienttopicmga_b", int(self._mqtt_mgatopic.get())) - cfg.set("mqttclienttopickey_b", int(self._mqtt_keytopic.get())) - cfg.set("mqttclienttopicfreq_b", int(self._mqtt_freqtopic.get())) - cfg.set("spartndecode_b", int(self._spartndecode.get())) - except (ValueError, TclError): - pass - - def _get_settings(self): - """ - Get settings from SPARTN handler. - """ - - self._connected = self.__app.spartn_handler.connected - if self._connected: - # get settings from running instance - self._settings = self.__app.spartn_handler.settings - else: - # get settings from saved configuration - cfg = self.__app.configuration - self._settings = {} - self._settings["server"] = cfg.get("mqttclientserver_s") - self._settings["port"] = cfg.get("mqttclientport_n") - self._settings["clientid"] = cfg.get("mqttclientid_s") - self._settings["region"] = cfg.get("mqttclientregion_s") - self._settings["mode"] = cfg.get("mqttclientmode_n") - self._settings["topic_ip"] = cfg.get("mqttclienttopicip_b") - self._settings["topic_mga"] = cfg.get("mqttclienttopicmga_b") - self._settings["topic_key"] = cfg.get("mqttclienttopickey_b") - self._settings["topic_freq"] = cfg.get("mqttclienttopicfreq_b") - self._settings["tlscrt"] = cfg.get("mqttclienttlscrt_s") - self._settings["tlskey"] = cfg.get("mqttclienttlskey_s") - self._settings["spartndecode"] = cfg.get("spartndecode_b") - self._settings["spartnkey"] = cfg.get("spartnkey_s") - if self._settings["spartnkey"] == "" or not HASCRYPTO: - self._settings["spartndecode"] = 0 - # if basedate is provided in config file, it must be an integer gnssTimetag - basedate = cfg.get("spartnbasedate_n") - if basedate == SPARTN_BASEDATE_CURRENT: - # pyspartn will interpret 'None' as current datetime - basedate = None - elif basedate == SPARTN_BASEDATE_DATASTREAM: - # pyspartn will interpret 'TIMEBASE' as use gnssTimeTag in datastream - basedate = TIMEBASE - self._settings["spartnbasedate"] = basedate - - self._mqtt_server.set(self._settings["server"]) - self._mqtt_port.set(self._settings["port"]) - self._mqtt_region.set(self._settings["region"]) - self._mqtt_source.set(self._settings.get("mode")) - self._mqtt_clientid.set(self._settings["clientid"]) - self._mqtt_iptopic.set(self._settings["topic_ip"]) - self._mqtt_mgatopic.set(self._settings["topic_mga"]) - self._mqtt_keytopic.set(self._settings["topic_key"]) - self._mqtt_freqtopic.set(self._settings["topic_freq"]) - self._mqtt_crt.set(self._settings["tlscrt"]) - self._mqtt_pem.set(self._settings["tlskey"]) - self._spartndecode.set(self._settings["spartndecode"]) - - def _set_settings(self): - """ - Set settings for SPARTN handler. - """ - - self._settings["server"] = self._mqtt_server.get() - self._settings["port"] = self._mqtt_port.get() - self._settings["region"] = self._mqtt_region.get() - self._settings["mode"] = self._mqtt_source.get() - self._settings["clientid"] = self._mqtt_clientid.get() - self._settings["topic_ip"] = self._mqtt_iptopic.get() - self._settings["topic_mga"] = self._mqtt_mgatopic.get() - self._settings["topic_key"] = self._mqtt_keytopic.get() - self._settings["topic_freq"] = self._mqtt_freqtopic.get() - self._settings["tlscrt"] = self._mqtt_crt.get() - self._settings["tlskey"] = self._mqtt_pem.get() - self._settings["spartndecode"] = self._spartndecode.get() - self._settings["output"] = self._output - - def set_controls(self, status: int): - """ - Enable or disable MQTT Correction widgets depending on - connection status. - - :param int status: connection status (0 = disconnected, 1 = connected) - """ - - stat = NORMAL if status == CONNECTED_SPARTNIP else DISABLED - for wdg in (self._btn_disconnect,): - wdg["state"] = stat - stat = DISABLED if status == CONNECTED_SPARTNIP else NORMAL - for wdg in ( - self._ent_mqttserver, - self._ent_mqttport, - self._btn_connect, - self._ent_mqttclientid, - self._ent_mqttcrt, - self._ent_mqttpem, - self._rad_ip, - self._rad_lb, - self._chk_mqtt_iptopic, - self._chk_mqtt_mgatopic, - self._chk_mqtt_keytopic, - self._chk_mqtt_freqtopic, - self._btn_opencrt, - self._btn_openpem, - ): - wdg["state"] = stat - stat = DISABLED if status == CONNECTED_SPARTNIP else READONLY - for wdg in (self._spn_mqttregion,): - wdg["state"] = stat - - if status == DISCONNECTED: - self._on_source(None, None, TRACEMODE_WRITE) - - def _valid_settings(self) -> bool: - """ - Validate settings. - - :return: valid True/False - :rtype: bool - """ - - valid = True - valid = valid & self._ent_mqttserver.validate(VALLEN, 1, 50) - valid = valid & self._ent_mqttport.validate(VALINT, 1, MAXPORT) - valid = valid & self._ent_mqttclientid.validate(VALLEN, 1, 50) - valid = valid & self._ent_mqttcrt.validate(VALLEN, 1, 254) - valid = valid & self._ent_mqttpem.validate(VALLEN, 1, 254) - return valid - - def _get_spartncerts(self, ext: str): - """ - Get SPARTN TLS key (*.pem) and cert (*.crt) files - - :param str ext: file extension ("crt" or "pem") - """ - - spfile = self.__app.file_handler.open_file( - self, - "spartncert", - ( - ("spartn files", f"*.{ext}"), - ("all files", "*.*"), - ), - ) - if ext == "crt": - self._mqtt_crt.set(spfile) - elif ext == "pem": - self._mqtt_pem.set(spfile) - - def on_connect(self): - """ - Connect to MQTT client. - """ - - if self.__app.rtk_conn_status == CONNECTED_SPARTNLB: - self.__container.status_label = ( - DLGSPARTNWARN.format("L-Band", "IP"), - ERRCOL, - ) - return - - server = "" - - if self._valid_settings(): - # if subscribing to SPARTNKEY topic, set pending flag - # for GNSS dialog to update key details when they arrive - # via RXM-SPARTNKEY message - if self._settings["topic_key"]: - self.__container.set_pending(RXMMSG, SPARTN_GNSS) - - self._set_settings() - # verbosity and logtofile set in App.__init__() - self.__app.spartn_handler.start( - server=self._settings["server"], - port=self._settings["port"], - clientid=self._settings["clientid"], - region=self._settings["region"], - mode=self._settings["mode"], - topic_ip=self._settings["topic_ip"], - topic_mga=self._settings["topic_mga"], - topic_key=self._settings["topic_key"], - topic_freq=self._settings["topic_freq"], - tlscrt=self._settings["tlscrt"], - tlskey=self._settings["tlskey"], - spartndecode=self._settings["spartndecode"], - spartnkey=self._settings["spartnkey"], - spartnbasedate=self._settings["spartnbasedate"], - output=self._settings["output"], - ) - self.set_controls(CONNECTED_SPARTNIP) - self.__container.status_label = (MQTTCONN.format(server), OKCOL) - self.__app.rtk_conn_status = CONNECTED_SPARTNIP - else: - self.__container.status_label = ("ERROR! Invalid Settings", ERRCOL) - - def on_disconnect(self, msg: str = ""): - """ - Disconnect from MQTT client. - - :param str msg: optional disconnection message - """ - - msg += "Disconnected" - if self.__app.rtk_conn_status == CONNECTED_SPARTNIP: - self.__app.spartn_handler.stop() - self.__app.rtk_conn_status = DISCONNECTED - self.__container.status_label = (msg, ERRCOL) - self.set_controls(DISCONNECTED) - - def update_status(self, msg: UBXMessage): - """ - Update pending confirmation status. - :param UBXMessage msg: UBX config message - """ - - @property - def clientid(self) -> str: - """ - Getter for Client ID. - - :return: client ID - :rtype: str - """ - - return self._mqtt_clientid.get() - - @clientid.setter - def clientid(self, clientid: str): - """ - Setter for Client ID. - - :param str clientid: Client ID - """ - - self._mqtt_clientid.set(clientid) - self._reset_keypaths(clientid) diff --git a/src/pygpsclient/stream_handler.py b/src/pygpsclient/stream_handler.py index c66156a9..1d1b6d6b 100644 --- a/src/pygpsclient/stream_handler.py +++ b/src/pygpsclient/stream_handler.py @@ -58,7 +58,6 @@ class to read and parse incoming data from the receiver. It places UNI_PROTOCOL, GNSSError, GNSSReader, - MQTTMessage, ) try: @@ -93,7 +92,6 @@ class to read and parse incoming data from the receiver. It places CONNECTED_SOCKET, DEFAULT_BUFSIZE, GNSS_PROTOCOL, - MQTT_PROTOCOL, SPARTN_PROTOCOL, STREAMDELAY, TTY_PROTOCOL, @@ -504,8 +502,6 @@ def _process_message( msgprot = UNI_PROTOCOL elif isinstance(parsed_data, SPARTNMessage) and protfilter & SPARTN_PROTOCOL: msgprot = SPARTN_PROTOCOL - elif isinstance(parsed_data, MQTTMessage): - msgprot = MQTT_PROTOCOL elif isinstance(parsed_data, GNSSMessage): msgprot = GNSS_PROTOCOL elif isinstance(parsed_data, str): @@ -531,7 +527,3 @@ def _process_message( # update GPX track file if enabled if self.__app.configuration.get("recordtrack_b"): self.__app.file_handler.update_gpx_track() - - # update database if enabled - if self.__app.configuration.get("database_b"): - self.__app.sqlite_handler.load_data() diff --git a/src/pygpsclient/strings.py b/src/pygpsclient/strings.py index d4315d9b..e86805f7 100644 --- a/src/pygpsclient/strings.py +++ b/src/pygpsclient/strings.py @@ -233,7 +233,6 @@ DLGTRINEX = "RINEX Conversion (EXPERIMENTAL)" DLGTSERVER = "Server Configuration" DLGTSETTINGS = "Settings" -DLGTSPARTN = "SPARTN Configuration (DEPRECATED)" DLGTTTY = "TTY Configuration" DLGTUBX = "UBX Configuration" DLGTUBXLEGACY = "UBX Legacy Configuration" diff --git a/src/pygpsclient/ubx_handler.py b/src/pygpsclient/ubx_handler.py index 4a7fc541..c709fee8 100644 --- a/src/pygpsclient/ubx_handler.py +++ b/src/pygpsclient/ubx_handler.py @@ -22,7 +22,7 @@ from pygpsclient.globals import BSR, GLONASS_NMEA, UTF8 from pygpsclient.helpers import corrage2int, fix2desc, hdg2yaw, ned2vector, svid2gnssid -from pygpsclient.strings import DLGTSERVER, DLGTSPARTN, DLGTUBX, DLGTUBXLEGACY, NA +from pygpsclient.strings import DLGTSERVER, DLGTUBX, DLGTUBXLEGACY, NA from pygpsclient.widget_state import VISIBLE, WDGSIGNALS, WDGSPECTRUM, WDGSYSMON UBXMODELS = { @@ -115,10 +115,6 @@ def process_data(self, raw_data: bytes, parsed_data: object): self._process_MONVER(parsed_data) elif parsed_data.identity == "RXM-RTCM": self._process_RXM_RTCM(parsed_data) - elif parsed_data.identity == "RXM-PMP": - self._process_RXM_PMP(parsed_data) - elif parsed_data.identity == "RXM-SPARTN-KEY": - self._process_RXM_SPARTN_KEY(parsed_data) def _process_ACK(self, msg: UBXMessage): """ @@ -132,10 +128,6 @@ def _process_ACK(self, msg: UBXMessage): if self.__app.dialog(DLGTUBXLEGACY) is not None: self.__app.dialog(DLGTUBXLEGACY).update_pending(msg) - # if SPARTN config dialog is open, send CFG & ACKs there - if self.__app.dialog(DLGTSPARTN) is not None: - self.__app.dialog(DLGTSPARTN).update_pending(msg) - # if Spectrumview, Sysmon or Signals widgets are active, send ACKSs there if msg.identity in ("ACK-ACK", "ACK-NAK"): wdgs = self.__app.widget_state.state @@ -677,26 +669,6 @@ def _process_MON_SPAN(self, data: UBXMessage): self.__app.gnss_status.spectrum_data = rfbs - def _process_RXM_SPARTN_KEY(self, data: UBXMessage): - """ - Process RXM-SPARTN_KEY sentences - poll response. - - :param UBXMessage data: RXM-SPARTN_KEY poll response message - """ - - if self.__app.dialog(DLGTSPARTN) is not None: - self.__app.dialog(DLGTSPARTN).update_pending(data) - - def _process_RXM_PMP(self, data: UBXMessage): - """ - Process RXM-PMP sentence - SPARTN L-Band data. - - :param UBXMessage data: RXM-PMP message - """ - - if self.__app.dialog(DLGTSPARTN) is not None: - self.__app.dialog(DLGTSPARTN).update_pending(data) - def _process_ESF_ALG(self, data: UBXMessage): """ Process ESF-ALG sentence - External Sensor Fusion IMU Alignment. diff --git a/tests/test_static.py b/tests/test_static.py index f40a8832..3482a497 100644 --- a/tests/test_static.py +++ b/tests/test_static.py @@ -134,18 +134,13 @@ def testconfiguration(self): cfg = Configuration(DummyApp()) self.assertEqual(cfg.get("bpsrate_n"), 9600) - self.assertEqual(cfg.get("lbandclientdrat_n"), 2400) self.assertEqual(cfg.get("userport_s"), "") - self.assertEqual(cfg.get("spartnport_s"), "") - self.assertEqual(len(cfg.settings), 159) - kwargs = {"userport": "/dev/ttyACM0", "spartnport": "/dev/ttyACM1"} + self.assertEqual(len(cfg.settings), 125) + kwargs = {"userport": "/dev/ttyACM0"} cfg.loadcli(**kwargs) self.assertEqual(cfg.get("userport_s"), "/dev/ttyACM0") - self.assertEqual(cfg.get("spartnport_s"), "/dev/ttyACM1") cfg.set("userport_s", "/dev/ttyACM2") - cfg.set("spartnport_s", "/dev/ttyACM3") self.assertEqual(cfg.get("userport_s"), "/dev/ttyACM2") - self.assertEqual(cfg.get("spartnport_s"), "/dev/ttyACM3") def testloadfile(self):