Coverage for .tox/py312/lib/python3.12/site-packages/pydalec/cli.py: 81%
104 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-04 16:06 +0200
1"""Command-line interface for connecting to a DALEC instrument over TCP."""
3from __future__ import annotations
5import argparse
6import sys
7import time
8from collections.abc import Sequence
10from pydalec.errors import (
11 PyDalecConnectionError,
12 PyDalecNoPositionDataError,
13 PyDalecNoSolarZenithDataError,
14)
15from pydalec.instrument import Dalec, Location
16from pydalec.logging_utils import enable_debug_logging
19class _CliArgumentError(ValueError):
20 """Raised when parsed CLI arguments violate local validation rules."""
23def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
24 """Parse CLI arguments for TCP connection and stream polling settings."""
25 parser = argparse.ArgumentParser(
26 prog='pydalec-test',
27 description='Connect to a DALEC instrument over TCP and stream measurements.',
28 )
29 parser.add_argument('ip', help='Instrument IP address')
30 parser.add_argument('--port', type=int, default=23, help='TCP port (default: 23)')
31 parser.add_argument(
32 '--data-root-dir',
33 type=str,
34 default=None,
35 help='Root directory where incoming data files are stored',
36 )
37 parser.add_argument(
38 '--max-file-size-kb',
39 type=int,
40 default=51200,
41 help='Maximum size in kB for each output file (default: 51200, i.e. 50 MB)',
42 )
43 parser.add_argument(
44 '--poll-interval',
45 type=float,
46 default=0.2,
47 help='Polling interval in seconds for new measurements (default: 0.2)',
48 )
49 parser.add_argument(
50 '--debug',
51 action='store_true',
52 help='Enable pydalec debug logging (without configuring the root logger)',
53 )
54 return parser.parse_args(argv)
57def _print_new_measurements(client: Dalec, last_measurement: object | None) -> object | None:
58 """Print newly received measurements and return the latest seen record."""
59 measurements = client.measurement_log
60 if not measurements:
61 return last_measurement
63 start_index = 0
64 if last_measurement is not None:
65 for index, measurement in enumerate(measurements):
66 if measurement == last_measurement:
67 start_index = index + 1
69 new_measurements = measurements[start_index:]
70 for measurement in new_measurements:
71 print(measurement)
73 return measurements[-1]
76def print_header(client: Dalec) -> None:
77 """Print the current location and sun zenith before streaming."""
78 try:
79 location: Location = client.get_location()
80 solar_zenith_deg: float = client.get_solar_zenith()
81 print(f'Current location: {location}')
82 print(f'Current sun zenith: {solar_zenith_deg} deg')
83 except PyDalecNoPositionDataError:
84 print('ERROR: No location data available')
85 except PyDalecNoSolarZenithDataError:
86 print('ERROR: No sun zenith data available')
87 else:
88 _ = input('Press Enter to start streaming measurements...')
91def _get_and_check_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
92 """Parse CLI arguments and check for required dependencies."""
93 args = _parse_args(argv)
94 raw_argv = list(argv) if argv is not None else sys.argv[1:]
96 if args.data_root_dir is None and '--max-file-size-kb' in raw_argv:
97 err_msg = '--max-file-size-kb requires --data-root-dir.'
98 raise _CliArgumentError(err_msg)
100 return args
103def _get_args_or_exit_code(argv: Sequence[str] | None = None) -> argparse.Namespace | int:
104 """Return parsed CLI arguments or an exit code for invalid invocation."""
105 try:
106 return _get_and_check_args(argv)
107 except _CliArgumentError as exc:
108 print(exc, file=sys.stderr)
109 return 2
110 except SystemExit as exc:
111 if exc.code in (None, 0):
112 return 0
113 if isinstance(exc.code, int):
114 return exc.code
115 print(exc.code, file=sys.stderr)
116 return 2
119def run(argv: Sequence[str] | None = None) -> int:
120 """Run the CLI workflow and return a shell-style exit code."""
121 args = _get_args_or_exit_code(argv)
122 if isinstance(args, int):
123 return args
125 if args.debug:
126 enable_debug_logging()
128 try:
129 client = Dalec.connect_tcp(
130 args.ip,
131 args.port,
132 data_root_dir=args.data_root_dir,
133 max_file_size_kb=args.max_file_size_kb,
134 )
135 except PyDalecConnectionError as exc:
136 print(f'Connection failed: {exc}', file=sys.stderr)
137 return 1
138 except ValueError as exc:
139 print(f'Invalid configuration: {exc}', file=sys.stderr)
140 return 2
142 print(f'Connected to DALEC at {args.ip}:{args.port}')
143 print_header(client)
145 exit_code = make_measurements(args, client)
147 return exit_code
150def make_measurements(args: argparse.Namespace, client: Dalec) -> int:
151 """Stream measurements until interrupted and return an exit code."""
152 last_measurement = None
153 exit_code = 0
154 measurements_started = False
155 try:
156 client.start_measurements()
157 measurements_started = True
158 print('Started measurements. Press Ctrl+C to stop.')
160 while True:
161 last_measurement = _print_new_measurements(client, last_measurement)
162 time.sleep(args.poll_interval)
163 except KeyboardInterrupt:
164 print('\nStopping measurement stream...')
165 except Exception as exc: # pragma: no cover
166 print(f'Runtime error: {exc}', file=sys.stderr)
167 exit_code = 1
168 finally:
169 if measurements_started:
170 try:
171 client.stop_measurements()
172 except Exception as exc: # pragma: no cover
173 print(f'Failed to stop measurements cleanly: {exc}', file=sys.stderr)
174 exit_code = 1
176 try:
177 client.disconnect()
178 except Exception as exc: # pragma: no cover
179 print(f'Failed to disconnect cleanly: {exc}', file=sys.stderr)
180 exit_code = 1
182 print('Disconnected.')
183 return exit_code
186def main(argv: Sequence[str] | None = None) -> None:
187 """Run the CLI and exit the process with the returned status code."""
188 raise SystemExit(run(argv))
191if __name__ == '__main__':
192 main()