-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodule_profiles.py
More file actions
156 lines (116 loc) · 4.8 KB
/
Copy pathmodule_profiles.py
File metadata and controls
156 lines (116 loc) · 4.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# module_profiles.py - GNSS module profile abstraction for multi-module support
from abc import ABC, abstractmethod
from typing import Optional
class ModuleProfile(ABC):
"""Abstract base class for GNSS module profiles.
Each profile encapsulates the proprietary command set and response parsing
logic for a specific GNSS module family.
"""
@property
@abstractmethod
def name(self) -> str:
"""Machine-readable profile identifier (e.g. 'lc29h')."""
@property
@abstractmethod
def display_name(self) -> str:
"""Human-readable module name (e.g. 'Quectel LC29H')."""
@abstractmethod
def firmware_command(self) -> Optional[str]:
"""Return the NMEA command to query firmware version, or None if unsupported."""
@abstractmethod
def parse_firmware_response(self, response: str) -> Optional[str]:
"""Extract the firmware version string from the module's response.
Returns the parsed version string, or None if parsing fails.
"""
@abstractmethod
def config_commands(self) -> list[dict]:
"""Return a list of configuration command dicts.
Each dict has:
- 'cmd': the NMEA command string (without $ prefix or checksum)
- 'ack': whether an ACK response is expected
"""
@abstractmethod
def check_ack(self, command: str, response: str) -> bool:
"""Check whether the response is a valid ACK for the given command.
Args:
command: The command that was sent (e.g. 'PAIR062,0,1').
response: The raw response line from the module.
Returns:
True if the response is a successful ACK.
"""
class LC29HProfile(ModuleProfile):
"""Profile for Quectel LC29H (DA) GNSS modules.
Uses PQTM commands for firmware queries and PAIR commands for configuration.
"""
@property
def name(self) -> str:
return "lc29h"
@property
def display_name(self) -> str:
return "Quectel LC29H"
def firmware_command(self) -> Optional[str]:
return "PQTMVERNO"
def parse_firmware_response(self, response: str) -> Optional[str]:
if not response or not response.startswith("$PQTMVERNO,"):
return None
parts = response.split(",")
if len(parts) >= 2 and parts[1]:
# Strip checksum suffix if present (e.g. "V1.0*3A")
version = parts[1].split("*")[0]
return version if version else None
return None
def config_commands(self) -> list[dict]:
return [
{"cmd": "PAIR062,0,1", "ack": True}, # GGA 1Hz
{"cmd": "PAIR062,4,1", "ack": True}, # RMC 1Hz
{"cmd": "PAIR062,2,1", "ack": True}, # GSA 1Hz
{"cmd": "PAIR062,3,1", "ack": True}, # GSV 1Hz
{"cmd": "PAIR062,5,1", "ack": True}, # VTG 1Hz
{"cmd": "PAIR436,1", "ack": True}, # Enable RTCM input
{"cmd": "PAIR513", "ack": True}, # Enable RTK mode
]
def check_ack(self, command: str, response: str) -> bool:
if not response:
return False
# Extract the command ID: strip the "PAIR" prefix to get the numeric part
cmd_name = command.split(",")[0]
cmd_id = cmd_name[4:] if cmd_name.startswith("PAIR") else cmd_name
ack_prefix = f"$PAIR001,{cmd_id},0"
return response.startswith(ack_prefix)
class GenericProfile(ModuleProfile):
"""Generic fallback profile for unknown GNSS modules.
Does not send any proprietary commands. Useful for modules that only
speak standard NMEA without vendor extensions.
"""
@property
def name(self) -> str:
return "generic"
@property
def display_name(self) -> str:
return "Generic GNSS"
def firmware_command(self) -> Optional[str]:
return None
def parse_firmware_response(self, response: str) -> Optional[str]:
return None
def config_commands(self) -> list[dict]:
return []
def check_ack(self, command: str, response: str) -> bool:
# Generic modules have no proprietary ACK mechanism
return True
# ---------------------------------------------------------------------------
# Profile registry
# ---------------------------------------------------------------------------
_PROFILE_REGISTRY: dict[str, type[ModuleProfile]] = {
"lc29h": LC29HProfile,
"generic": GenericProfile,
}
def get_profile(module_name: str) -> ModuleProfile:
"""Look up a module profile by name (case-insensitive).
Falls back to GenericProfile for unrecognised names.
"""
key = module_name.strip().lower()
profile_cls = _PROFILE_REGISTRY.get(key, GenericProfile)
return profile_cls()
def list_profiles() -> list[str]:
"""Return sorted list of registered profile names."""
return sorted(_PROFILE_REGISTRY.keys())