[stn] added resection calc and station upload commands

Author: MrClock8163

src/instrumentman/__init__.py

@@ -9,6 +9,7 @@
 from . import terminal
 from . import setup
 from . import setmeasurement
+from . import resection
 from . import protocoltest
 from . import inclination
 from . import filetransfer
@@ -77,6 +78,7 @@ def cli_upload() -> None:
 cli_measure.add_command(inclination.cli_measure)
 cli_calc.add_command(setmeasurement.cli_calc)
 cli_calc.add_command(inclination.cli_calc)
+cli_calc.add_command(resection.cli_calc)
 cli_test.add_command(protocoltest.cli_geocom)
 cli_test.add_command(protocoltest.cli_gsidna)
 cli_merge.add_command(setmeasurement.cli_merge)
@@ -91,3 +93,4 @@ def cli_upload() -> None:
 cli_download.add_command(settings.cli_download)
 cli_upload.add_command(datatransfer.cli_upload)
 cli_upload.add_command(settings.cli_upload)
+cli_upload.add_command(resection.cli_upload)

src/instrumentman/resection/__init__.py

@@ -0,0 +1,85 @@
+from typing import Any
+
+from click_extra import (
+    extra_command,
+    argument,
+    option,
+    file_path
+)
+
+from ..utils import (
+    com_port_argument,
+    com_option_group,
+    Angle
+)
+
+
+@extra_command(
+    "station",
+    params=None,
+    context_settings={"auto_envvar_prefix": None}
+)  # type: ignore[misc]
+@argument(
+    "measurements",
+    help="input session file to process",
+    type=file_path(exists=True)
+)
+@argument(
+    "targets",
+    type=file_path(exists=True),
+    help="JSON file containing target definitions"
+)
+@argument(
+    "output",
+    help="output JSON file",
+    type=file_path(readable=False)
+)
+@option(
+    "--points",
+    help="target points to use as references",
+    type=str
+)
+@option(
+    "--height",
+    help="instrument height",
+    type=float,
+    default=0
+)
+def cli_calc(**kwargs: Any) -> None:
+    """Calculate station coordinates from set measurements by resection."""
+    from .calculate import main
+
+    main(**kwargs)
+
+
+@extra_command(
+    "station",
+    params=None,
+    context_settings={"auto_envvar_prefix": None}
+)  # type: ignore[misc]
+@com_port_argument()
+@com_option_group()
+@option(
+    "--coordinates",
+    help="station coordinates",
+    type=(float, float, float),
+    default=(0, 0, 0)
+)
+@option(
+    "--instrumentheight",
+    "--iheight",
+    help="instrument height",
+    type=float,
+    default=0
+)
+@option(
+    "--orientation",
+    help="instrument orientation correction",
+    type=Angle(),
+    default="0-00-00"
+)
+def cli_upload(**kwargs: Any) -> None:
+    """Upload station setup to instrument."""
+    from .upload import main
+
+    main(**kwargs)

src/instrumentman/resection/calculate.py

@@ -0,0 +1,111 @@
+from pathlib import Path
+import json
+
+from jsonschema import ValidationError
+from geocompy.data import Angle, Coordinate
+
+from ..targets import load_targets_from_json
+from ..setmeasurement.sessions import SessionDict
+from ..setmeasurement.process import (
+    SessionValidator,
+    calc_angles
+)
+from ..utils import echo_red
+from ..calculations import resection_2d_1d, preliminary_resection
+
+
+def main(
+    measurements: Path,
+    targets: Path,
+    output: Path,
+    points: str | None = None,
+    height: float = 0
+) -> None:
+    with measurements.open("rt", encoding="utf8") as file:
+        data: SessionDict = json.load(file)
+
+    validator = SessionValidator(False)
+    try:
+        validator.validate(data)
+    except ValidationError as ve:
+        echo_red("Measurement data does not follow the required schema")
+        echo_red(ve)
+        exit(4)
+    except ValueError as e:
+        echo_red("The Measurement data did not pass validation")
+        echo_red(e)
+        exit(4)
+
+    try:
+        targetlist = load_targets_from_json(str(targets))
+    except ValidationError:
+        echo_red("Target data does not follow the required schema")
+        exit(4)
+
+    target_names = targetlist.get_target_names()
+    if points is not None:
+        target_names = list(set(target_names).intersection(points.split(",")))
+
+    actual_targets: list[str] = []
+    references: list[Coordinate] = []
+    obs: list[tuple[Angle, Angle, float]] = []
+    for cycle in data["cycles"]:
+        for p in cycle["points"]:
+            name = p["name"]
+            if name not in target_names:
+                continue
+
+            actual_targets.append(name)
+
+            t = targetlist.get_target(name)
+            coord = t.coords
+            coord.z += t.height
+            references.append(coord)
+            hz = Angle(p["face1"][0])
+            v = Angle(p["face1"][1])
+            d = p["face1"][2]
+            if p.get("face2") is not None:
+                hz, v, _, _ = calc_angles(
+                    hz,
+                    v,
+                    Angle(p["face2"][0]),
+                    Angle(p["face2"][1])
+                )
+                d = (d + p["face2"][2]) / 2
+
+            obs.append(
+                (hz, v, d)
+            )
+
+    if len(set(actual_targets)) < 2:
+        echo_red("Cannot calculate resection from less than 2 targets")
+        exit(1)
+
+    (
+        converged,
+        orientation,
+        stdev_orientation,
+        station,
+        stdev_station
+    ) = resection_2d_1d(
+        obs,
+        references,
+        preliminary_resection(
+            obs,
+            references
+        )
+    )
+
+    if not converged:
+        echo_red("Resection calculation failed")
+        exit(1)
+
+    results = {
+        "station": list(station - Coordinate(0, 0, height)),
+        "stddev_station": list(stdev_station),
+        "instrumentheight": height,
+        "orientation": orientation.to_dms(4),
+        "stddev_orientation": stdev_orientation.to_dms(4)
+    }
+    with output.open("wt", encoding="utf8") as file:
+        json.dump(results, file, indent=4)

src/instrumentman/resection/upload.py

@@ -0,0 +1,46 @@
+from geocompy.data import Angle, Coordinate
+from geocompy.communication import open_serial
+from geocompy.geo import GeoCom
+from geocompy.geo.gctypes import GeoComCode
+
+from ..utils import echo_red, echo_green
+
+
+def main(
+    port: str,
+    baud: int = 9600,
+    timeout: int = 15,
+    retry: int = 1,
+    sync_after_timeout: bool = False,
+    coordinates: tuple[float, float, float] = (0, 0, 0),
+    instrumentheight: float = 0,
+    orientation: str = "0-00-00"
+) -> None:
+    station = Coordinate(*coordinates)
+    ori = Angle.from_dms(orientation)
+    with open_serial(
+        port=port,
+        speed=baud,
+        timeout=timeout,
+        retry=retry,
+        sync_after_timeout=sync_after_timeout
+    ) as com:
+        tps = GeoCom(com)
+        resp_stn = tps.tmc.set_station(station, instrumentheight)
+        if resp_stn.error != GeoComCode.OK:
+            echo_red("Cannot set station")
+            exit(1)
+
+        resp_angle = tps.tmc.get_angle()
+        if resp_angle.error != GeoComCode.OK or resp_angle.params is None:
+            echo_red("Cannot set orientation")
+            exit(1)
+
+        resp_ori = tps.tmc.set_azimuth(
+            (resp_angle.params[0] + ori).normalized()
+        )
+        if resp_ori.error != GeoComCode.OK:
+            echo_red("Cannot set orientation")
+            exit(1)
+
+        echo_green("Station and orientation set")

src/instrumentman/setmeasurement/__init__.py

@@ -202,6 +202,7 @@ def cli_validate(**kwargs: Any) -> None:
     type=(float, float, float)
 )
 @option(
+    "--instrumentheight",
     "--iheight",
     help="override instrument height",
     type=float
@@ -213,7 +214,7 @@ def cli_validate(**kwargs: Any) -> None:
 )
 @constraint(
     all_or_none,
-    ["station", "iheight"]
+    ["station", "instrumentheight"]
 )
 def cli_calc(**kwargs: Any) -> None:
     """Calculate results from set measurements.

src/instrumentman/setmeasurement/process.py

@@ -201,7 +201,7 @@ def main_calc(
     precision: int = 4,
     allow_oneface: bool = False,
     station: tuple[float, float, float] | None = None,
-    iheight: float | None = None,
+    instrumentheight: float | None = None,
     orientation: str | None = None
 ) -> None:
     with input.open("rt", encoding="utf8") as file:
@@ -222,7 +222,7 @@ def main_calc(
     points = {"points": search("cycles[].points[]", data)}
     ptids = list(set(search("points[].name", points)))
 
-    if station is None or iheight is None:
+    if station is None or instrumentheight is None:
         stn = Coordinate(
             *data["station"]
         ) + Coordinate(
@@ -234,7 +234,7 @@ def main_calc(
         stn = Coordinate(
             station[0],
             station[1],
-            station[2] + iheight
+            station[2] + instrumentheight
         )
 
     if orientation is None: