Updated VivaTPS to new geocom protocol

Author: MrClock8163

src/geocompy/protocols.py

@@ -174,69 +174,15 @@ def request(
         rpc: int,
         params: Iterable[int | float | bool | str | Angle | Byte] = (),
         parsers: Callable[[str], _T] | None = None
-    ) -> GeoComResponse[_T]:
-        """
-        Executes an RPC request and returns the parsed GeoCom response.
-
-        Constructs a request (from the given RPC code and parameters),
-        writes it to the serial line, then reads the response. The
-        response is then parsed using the provided parser function.
-
-        Parameters
-        ----------
-        rpc : int
-            Number of the RPC to execute.
-        params : Iterable[int | float | bool | str | Angle | Byte], optional
-            Parameters for the request, by default []
-        parsers : Callable[[str], Any] | None, optional
-            Parser function for the value in the RPC response,
-            by default None
-
-        Returns
-        -------
-        GeoComResponse
-            Parsed return codes and parameter from the RPC response.
-
-        Raises
-        ------
-        NotImplementedError
-            If the method is not implemented on the class.
-        """
+    ) -> GeoComResponse[_T]: ...
 
     @overload
     def request(
         self,
         rpc: int,
         params: Iterable[int | float | bool | str | Angle | Byte] = (),
         parsers: Iterable[Callable[[str], Any]] | None = None
-    ) -> GeoComResponse[tuple]:
-        """
-        Executes an RPC request and returns the parsed GeoCom response.
-
-        Constructs a request (from the given RPC code and parameters),
-        writes it to the serial line, then reads the response. The
-        response is then parsed using the provided parser functions.
-
-        Parameters
-        ----------
-        rpc : int
-            Number of the RPC to execute.
-        params : Iterable[int | float | bool | str | Angle | Byte], optional
-            Parameters for the request, by default ()
-        parsers : Iterable[Callable[[str], Any]] | None, optional
-            Parser functions for the values in the RPC response,
-            by default None
-
-        Returns
-        -------
-        GeoComResponse
-            Parsed return codes and parameters from the RPC response.
-
-        Raises
-        ------
-        NotImplementedError
-            If the method is not implemented on the class.
-        """
+    ) -> GeoComResponse[tuple]: ...
 
     def request(
         self,
@@ -284,63 +230,15 @@ def parse_response(
         cmd: str,
         response: str,
         parsers: Callable[[str], _T] | None = None
-    ) -> GeoComResponse[_T]:
-        """
-        Parses RPC response and constructs :class:`GeoComResponse`
-        instance.
-
-        Parameters
-        ----------
-        cmd : str
-            Full, serialized request, that invoked the response.
-        response : str
-            Full, received response.
-        parsers : Callable[[str], Any] | None, optional
-            Parser function for the value in the RPC response, by
-            default None
-
-        Returns
-        -------
-        GeoComResponse
-            Parsed return codes and parameter from the RPC response.
-
-        Raises
-        ------
-        NotImplementedError
-            If the method is not implemented on the class.
-        """
+    ) -> GeoComResponse[_T]: ...
 
     @overload
     def parse_response(
         cls,
         cmd: str,
         response: str,
         parsers: Iterable[Callable[[str], Any]] | None = None
-    ) -> GeoComResponse[tuple]:
-        """
-        Parses RPC response and constructs :class:`GeoComResponse`
-        instance.
-
-        Parameters
-        ----------
-        cmd : str
-            Full, serialized request, that invoked the response.
-        response : str
-            Full, received response.
-        parsers : Iterable[Callable[[str], Any]], optional
-            Parser functions for the values in the RPC response, by
-            default None
-
-        Returns
-        -------
-        GeoComResponse
-            Parsed return codes and parameters from the RPC response.
-
-        Raises
-        ------
-        NotImplementedError
-            If the method is not implemented on the class.
-        """
+    ) -> GeoComResponse[tuple]: ...
 
     def parse_response(
         cls,

src/geocompy/vivatps/__init__.py

@@ -39,7 +39,7 @@
 import logging
 from time import sleep
 from traceback import format_exc
-from typing import Callable, Iterable, Any
+from typing import Callable, Iterable, Any, overload, TypeVar
 
 from serial import SerialException, SerialTimeoutException
 
@@ -69,6 +69,9 @@
 from .grc import VivaTPSGRC, rpcnames
 
 
+_T = TypeVar("_T")
+
+
 class VivaTPS(GeoComProtocol):
     """
     VivaTPS GeoCom protocol handler.
@@ -193,12 +196,17 @@ def __init__(
 
         self._precision = 15
         resp = self.get_double_precision()
-        if resp.comcode and resp.rpccode:
-            self._precision = resp.params["digits"]
+        if resp.params is not None:
+            self._precision = resp.params
+        else:
+            self._logger.error(
+                f"Could not syncronize double precision, "
+                f"defaulting to {self._precision:d}"
+            )
 
         self._logger.info("Connection initialized")
 
-    def get_double_precision(self) -> GeoComResponse:
+    def get_double_precision(self) -> GeoComResponse[int]:
         """
         RPC 108, ``COM_GetDoublePrecision``
 
@@ -218,13 +226,13 @@ def get_double_precision(self) -> GeoComResponse:
         """
         return self.request(
             108,
-            parsers={"digits": int}
+            parsers=int
         )
 
     def set_double_precision(
         self,
         digits: int
-    ) -> GeoComResponse:
+    ) -> GeoComResponse[None]:
         """
         RPC 107, ``COM_SetDoublePrecision``
 
@@ -245,16 +253,36 @@ def set_double_precision(
         get_double_precision
 
         """
-        response = self.request(107, [digits])
+        response: GeoComResponse[None] = self.request(107, [digits])
         if response.comcode and response.rpccode:
             self._precision = digits
         return response
 
+    @overload
+    def request(
+        self,
+        rpc: int,
+        params: Iterable[int | float | bool | str | Angle | Byte] = (),
+        parsers: Callable[[str], _T] | None = None
+    ) -> GeoComResponse[_T]: ...
+
+    @overload
+    def request(
+        self,
+        rpc: int,
+        params: Iterable[int | float | bool | str | Angle | Byte] = (),
+        parsers: Iterable[Callable[[str], Any]] | None = None
+    ) -> GeoComResponse[tuple]: ...
+
     def request(
         self,
         rpc: int,
         params: Iterable[int | float | bool | str | Angle | Byte] = [],
-        parsers: dict[str, Callable[[str], Any]] | None = None
+        parsers: (
+            Iterable[Callable[[str], Any]]
+            | Callable[[str], Any]
+            | None
+        ) = None
     ) -> GeoComResponse:
         """
         Executes a VivaTPS RPC request and returns the parsed GeoCom
@@ -377,16 +405,36 @@ def request(
         response = self.parse_response(
             cmd,
             answer,
-            parsers if parsers is not None else {}
+            parsers
         )
         self._logger.debug(response)
         return response
 
+    @overload
     def parse_response(
         self,
         cmd: str,
         response: str,
-        parsers: dict[str, Callable[[str], Any]]
+        parsers: Callable[[str], _T] | None = None
+    ) -> GeoComResponse[_T]: ...
+
+    @overload
+    def parse_response(
+        self,
+        cmd: str,
+        response: str,
+        parsers: Iterable[Callable[[str], Any]] | None = None
+    ) -> GeoComResponse[tuple]: ...
+
+    def parse_response(
+        self,
+        cmd: str,
+        response: str,
+        parsers: (
+            Iterable[Callable[[str], Any]]
+            | Callable[[str], Any]
+            | None
+        ) = None
     ) -> GeoComResponse:
         """
         Parses RPC response and constructs GeoComResponse
@@ -425,37 +473,49 @@ def parse_response(
                 response,
                 VivaTPSGRC.COM_CANT_DECODE,
                 VivaTPSGRC.UNDEFINED,
-                0,
-                {}
+                0
             )
 
         groups = m.groupdict()
         values = groups.get("params", "")
         if values is None:
             values = ""
-        params: dict = {}
+
+        if parsers is None:
+            parsers = ()
+        elif not isinstance(parsers, Iterable):
+            parsers = (parsers,)
+
+        params: list = []
         try:
-            for (name, func), value in zip(parsers.items(), values.split(",")):
-                params[name] = func(value)
+            for func, value in zip(parsers, values.split(",")):
+                params.append(func(value))
         except Exception:
             return GeoComResponse(
                 rpcname,
                 cmd,
                 response,
                 VivaTPSGRC.COM_CANT_DECODE,
                 VivaTPSGRC.UNDEFINED,
-                0,
-                {}
+                0
             )
 
         comrc = VivaTPSGRC(int(groups["comrc"]))
         rc = VivaTPSGRC(int(groups["rc"]))
+        match len(params):
+            case 0:
+                params_final = None
+            case 1:
+                params_final = params[1]
+            case _:
+                params_final = tuple(params)
+
         return GeoComResponse(
             rpcname,
             cmd,
             response,
             comrc,
             rc,
             int(groups["tr"]),
-            params
+            params_final
         )

src/geocompy/vivatps/aut.py

@@ -43,7 +43,7 @@ class ONOFF(Enum):
     def set_lock_fly_mode(
         self,
         state: ONOFF | str
-    ) -> GeoComResponse:
+    ) -> GeoComResponse[None]:
         """
         RPC 9103, ``AUT_SetLockFlyMode``
 
@@ -68,7 +68,7 @@ def set_lock_fly_mode(
             [_state.value]
         )
 
-    def get_lock_fly_mode(self) -> GeoComResponse:
+    def get_lock_fly_mode(self) -> GeoComResponse[ONOFF]:
         """
         RPC 9102, ``AUT_GetLockFlyMode``
 
@@ -86,17 +86,15 @@ def get_lock_fly_mode(self) -> GeoComResponse:
         """
         return self._request(
             9102,
-            parsers={
-                "state": enumparser(self.ONOFF)
-            }
+            parsers=enumparser(self.ONOFF)
         )
 
     def cam_posit_to_pixel_coord(
         self,
         x: int,
         y: int,
         camtype: VivaTPSCAM.CAMTYPE | str = VivaTPSCAM.CAMTYPE.OVC
-    ) -> GeoComResponse:
+    ) -> GeoComResponse[None]:
         """
         RPC 9081, ``AUT_CAM_PositToPixelCoord``
 

src/geocompy/vivatps/bap.py

@@ -37,7 +37,7 @@ class ONOFF(Enum):
         OFF = 0
         ON = 1
 
-    def get_atr_precise(self) -> GeoComResponse:
+    def get_atr_precise(self) -> GeoComResponse[ONOFF]:
         """
         RPC 17039, ``BAP_GetATRPrecise``
 
@@ -55,15 +55,13 @@ def get_atr_precise(self) -> GeoComResponse:
         """
         return self._request(
             17039,
-            parsers={
-                "state": enumparser(self.ONOFF)
-            }
+            parsers=enumparser(self.ONOFF)
         )
 
     def set_atr_precise(
         self,
         state: ONOFF | str
-    ) -> GeoComResponse:
+    ) -> GeoComResponse[None]:
         """
         RPC 17040, ``BAP_SetATRPrecise``