|
11 | 11 |
|
12 | 12 | import logging |
13 | 13 | import re |
14 | | -from collections.abc import Callable |
| 14 | +from codecs import BOM_UTF16_BE, BOM_UTF16_LE, BOM_UTF32_BE, BOM_UTF32_LE |
| 15 | +from collections.abc import Callable, Iterator |
| 16 | +from io import TextIOWrapper |
15 | 17 | from math import isfinite as math_isfinite |
16 | 18 | from os import path as os_path |
17 | 19 | from shutil import get_terminal_size |
@@ -41,6 +43,47 @@ class ParamFileError(ValueError): |
41 | 43 | """Raised when a .param file contains invalid or malformed data.""" |
42 | 44 |
|
43 | 45 |
|
| 46 | +def _has_zero_byte_pattern(sample: bytes, stride: int, zero_offsets: tuple[int, ...]) -> bool: |
| 47 | + """Return whether the sample matches zero-byte offsets for a fixed-width encoding.""" |
| 48 | + usable_length = len(sample) - len(sample) % stride |
| 49 | + return usable_length >= stride * 2 and all( |
| 50 | + sample[offset] == 0 for zero_offset in zero_offsets for offset in range(zero_offset, usable_length, stride) |
| 51 | + ) |
| 52 | + |
| 53 | + |
| 54 | +def _detect_param_file_encoding(sample: bytes) -> str: |
| 55 | + """Detect UTF-8/16/32 using a BOM or the zero-byte pattern of ASCII parameter text.""" |
| 56 | + for byte_order_mark, encoding in ( |
| 57 | + (BOM_UTF32_LE, "utf-32"), |
| 58 | + (BOM_UTF32_BE, "utf-32"), |
| 59 | + (BOM_UTF16_LE, "utf-16"), |
| 60 | + (BOM_UTF16_BE, "utf-16"), |
| 61 | + ): |
| 62 | + if sample.startswith(byte_order_mark): |
| 63 | + return encoding |
| 64 | + |
| 65 | + for encoding, stride, zero_offsets in ( |
| 66 | + ("utf-32-le", 4, (1, 2, 3)), |
| 67 | + ("utf-32-be", 4, (0, 1, 2)), |
| 68 | + ("utf-16-le", 2, (1,)), |
| 69 | + ("utf-16-be", 2, (0,)), |
| 70 | + ): |
| 71 | + if _has_zero_byte_pattern(sample, stride, zero_offsets): |
| 72 | + return encoding |
| 73 | + |
| 74 | + return "utf-8-sig" |
| 75 | + |
| 76 | + |
| 77 | +def read_param_file_lines(param_file: str) -> Iterator[str]: |
| 78 | + """Yield parameter-file lines from UTF-8/16/32 files, with or without BOMs.""" |
| 79 | + with open(param_file, "rb") as binary_handle: |
| 80 | + sample = binary_handle.read(256) |
| 81 | + binary_handle.seek(0) |
| 82 | + encoding = _detect_param_file_encoding(sample) |
| 83 | + with TextIOWrapper(binary_handle, encoding=encoding) as text_handle: |
| 84 | + yield from text_handle |
| 85 | + |
| 86 | + |
44 | 87 | def validate_param_name(param_name: str) -> tuple[bool, str]: |
45 | 88 | """ |
46 | 89 | Validate parameter name according to ArduPilot standards. |
@@ -151,37 +194,36 @@ def load_param_file_into_dict(param_file: str) -> "ParDict": |
151 | 194 | """ |
152 | 195 | parameter_dict = ParDict() |
153 | 196 | try: |
154 | | - with open(param_file, encoding="utf-8-sig") as f_handle: |
155 | | - for i, f_line in enumerate(f_handle, start=1): |
156 | | - original_line = f_line |
157 | | - line = f_line.strip() |
158 | | - comment = None |
159 | | - if not line: |
160 | | - continue # skip empty lines |
161 | | - if line[0] == "#": |
162 | | - continue # skip comments |
163 | | - if "#" in line: |
164 | | - line, comment = line.split("#", 1) # strip trailing comments |
165 | | - comment = comment.strip() |
166 | | - if "," in line: |
167 | | - # parse mission planner style parameter files |
168 | | - parameter, value = line.split(",", 1) |
169 | | - elif " " in line: |
170 | | - # parse mavproxy style parameter files |
171 | | - parameter, value = line.split(" ", 1) |
172 | | - elif "\t" in line: |
173 | | - parameter, value = line.split("\t", 1) |
174 | | - else: |
175 | | - msg = _("Missing parameter-value separator: {line} in {param_file} line {i}").format( |
176 | | - line=line, param_file=param_file, i=i |
177 | | - ) |
178 | | - raise ParamFileError(msg) |
179 | | - # Strip whitespace from both parameter name and value immediately after splitting |
180 | | - parameter = parameter.strip() |
181 | | - value = value.strip() |
182 | | - ParDict._validate_parameter(param_file, parameter_dict, i, original_line, comment, parameter, value) |
| 197 | + for i, f_line in enumerate(read_param_file_lines(param_file), start=1): |
| 198 | + original_line = f_line |
| 199 | + line = f_line.strip() |
| 200 | + comment = None |
| 201 | + if not line: |
| 202 | + continue # skip empty lines |
| 203 | + if line[0] == "#": |
| 204 | + continue # skip comments |
| 205 | + if "#" in line: |
| 206 | + line, comment = line.split("#", 1) # strip trailing comments |
| 207 | + comment = comment.strip() |
| 208 | + if "," in line: |
| 209 | + # parse mission planner style parameter files |
| 210 | + parameter, value = line.split(",", 1) |
| 211 | + elif " " in line: |
| 212 | + # parse mavproxy style parameter files |
| 213 | + parameter, value = line.split(" ", 1) |
| 214 | + elif "\t" in line: |
| 215 | + parameter, value = line.split("\t", 1) |
| 216 | + else: |
| 217 | + msg = _("Missing parameter-value separator: {line} in {param_file} line {i}").format( |
| 218 | + line=line, param_file=param_file, i=i |
| 219 | + ) |
| 220 | + raise ParamFileError(msg) |
| 221 | + # Strip whitespace from both parameter name and value immediately after splitting |
| 222 | + parameter = parameter.strip() |
| 223 | + value = value.strip() |
| 224 | + ParDict._validate_parameter(param_file, parameter_dict, i, original_line, comment, parameter, value) |
183 | 225 | except UnicodeDecodeError as exp: |
184 | | - msg = _("Fatal error reading {param_file}, file must be UTF-8 encoded: {exp}").format( |
| 226 | + msg = _("Fatal error reading {param_file}, file must be UTF-8, UTF-16, or UTF-32 encoded: {exp}").format( |
185 | 227 | param_file=param_file, exp=exp |
186 | 228 | ) |
187 | 229 | raise ParamFileError(msg) from exp |
|
0 commit comments