From 0d0852cf2657d8cd56a82a4844f8b07590274023 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:17:48 -0700 Subject: [PATCH 01/15] [#1669] Support CCSDS namespaces in OMM XML - Accept prefixed and default CCSDS namespaces. - Reject duplicate OMM fields and blocks while preserving record diagnostics. - Add namespace and collision regressions; update the beta feature release note. Validation: 874 tests passed, along with pre-commit and focused Sphinx checks. --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 192 +++++++++++++++++- src/utilities/ommHandling.py | 61 ++++-- 3 files changed, 232 insertions(+), 23 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index e0254c64a34..ae22fa8dff0 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 8bbe991c3d5..c20f450b6cb 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -22,6 +22,7 @@ # Creation Date: September 16, 2026 # +import copy import datetime as dt import json import xml.etree.ElementTree as ET @@ -109,13 +110,161 @@ def block(name, keys): + "" + block("meanElements", meanKeys) + block("tleParameters", tleKeys) + "" + "" ) - path.write_text(f"{''.join(segments)}", encoding="utf-8") + path.write_text(f"{''.join(segments)}", encoding="utf-8") + return path + + +def _set_xml_namespace(path, prefix, namespace="urn:ccsds:schema:ndmxml"): + """Qualify fixture elements without changing foreign tags or global prefix registrations.""" + if prefix is None: + return path + tree = ET.parse(path) + if prefix: + for element in tree.iter(): + if not element.tag.startswith("{"): + element.tag = f"{prefix}:{element.tag}" + tree.getroot().set(f"xmlns:{prefix}" if prefix else "xmlns", namespace) + tree.write(path, encoding="utf-8") return path _WRITERS = {"kvn": _writeKvn, "json": _writeJson, "csv": _writeCsv, "xml": _writeXml} +@pytest.mark.parametrize("namespace_prefix", ["ndm", "orbit", ""]) +def test_omm_xml_namespace_preserves_fields_and_orbit(tmp_path, capsys, namespace_prefix): + """CCSDS namespaces preserve all 21 fields, output metadata, and osculating elements.""" + reference_path = _writeXml(tmp_path / "reference.xml", [_OMM_FIELDS]) + path = _set_xml_namespace( + _writeXml(tmp_path / "qualified.xml", [_OMM_FIELDS]), namespace_prefix + ) + + assert ommHandling._parseOmmXml(path.read_text(encoding="utf-8")) == [_OMM_FIELDS] + records = ommHandling.satOmm2elem(str(path)) + reference = ommHandling.satOmm2elem(str(reference_path))[0] + + assert len(records) == 1 + for name in reference.__dataclass_fields__: + if name != "oe": + assert getattr(records[0], name) == getattr(reference, name) + for name in ("a", "e", "i", "Omega", "omega", "f"): + assert getattr(records[0].oe, name) == getattr(reference.oe, name) + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("qualified_first", [False, True]) +@pytest.mark.parametrize("element_path", [ + "metadata/OBJECT_NAME", "metadata/CENTER_NAME", "data/meanElements/INCLINATION", + "data/tleParameters/CLASSIFICATION_TYPE", "metadata", "data", "data/meanElements", + "data/tleParameters", +]) +def test_omm_xml_namespace_collisions_preserve_record_numbers( + tmp_path, monkeypatch, capsys, qualified_first, element_path +): + """Namespace aliases cannot hide ambiguous fields or blocks in either element order.""" + broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") + incomplete = {name: value for name, value in broken.items() if name != "MEAN_MOTION"} + second = dict(_OMM_FIELDS, OBJECT_NAME="SECOND-SAT", NORAD_CAT_ID="25545") + path = _writeXml(tmp_path / "collisions.xml", [broken, _OMM_FIELDS, incomplete, second]) + tree = ET.parse(path) + segment = tree.find(".//segment") + original = segment.find(element_path) + qualified = copy.deepcopy(original) + for element in qualified.iter(): + element.tag = f"{{urn:ccsds:schema:ndmxml}}{element.tag}" + if element_path.endswith("/INCLINATION"): + original.set("units", "rad") + qualified.set("units", "deg") + parent_path, _, name = element_path.rpartition("/") + parent = segment.find(parent_path) if parent_path else segment + position = list(parent).index(original) + parent.insert(position if qualified_first else position + 1, qualified) + tree.write(path, encoding="utf-8") + + expected_names = [_OMM_FIELDS["OBJECT_NAME"], second["OBJECT_NAME"]] + initialize = ommHandling.sgp4omm.initialize + initialized_names = [] + + def initialize_valid_record(satellite, fields): + """Fail if an ambiguous XML record reaches SGP4.""" + assert fields["OBJECT_NAME"] in expected_names + initialized_names.append(fields["OBJECT_NAME"]) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(str(path)) + + assert [record.satName for record in records] == expected_names + assert initialized_names == expected_names + warnings = capsys.readouterr().out.splitlines() + assert len(warnings) == 2 + assert f"skipped OMM record 1 in {path}" in warnings[0] + assert "Duplicate OMM XML element" in warnings[0] + assert name.upper() in warnings[0] + assert f"skipped OMM record 3 in {path}" in warnings[1] + assert "MEAN_MOTION" in warnings[1] + + +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +def test_omm_xml_repeated_comments_and_extensions_are_allowed(tmp_path, capsys, namespace_prefix): + """Repeated comments and foreign fields do not make an OMM record ambiguous.""" + path = _writeXml(tmp_path / "comments.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + for block_name in ("metadata", "meanElements", "tleParameters"): + block = tree.find(f".//{block_name}") + for text in ("FIRST", "SECOND"): + ET.SubElement(block, "COMMENT").text = text + ET.SubElement(block, "{urn:example:foreign}INCLINATION", {"units": "rad"}).text = text + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + + records = ommHandling.satOmm2elem(str(path)) + + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +@pytest.mark.parametrize("tag", [ + "segment", "metadata", "data", "meanElements", "tleParameters", "OBJECT_NAME", "MEAN_MOTION", +]) +def test_omm_xml_foreign_namespace_cannot_override_records(tmp_path, capsys, namespace_prefix, tag): + """Foreign elements with matching local names cannot replace OMM blocks or values.""" + path = _writeXml(tmp_path / "foreign-elements.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + original = tree.find(f".//{tag}") + foreign = copy.deepcopy(original) + for element in foreign.iter(): + element.tag = f"{{urn:example:foreign}}{element.tag}" + if len(element) == 0: + element.text = "FOREIGN" + parent = tree.find(f".//{tag}/..") + # Structural collisions precede valid blocks; field collisions follow valid values. + parent.insert(0 if len(original) else len(parent), foreign) + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + + raw_records = ommHandling._parseOmmXml(path.read_text(encoding="utf-8")) + assert len(raw_records) == 1 + assert {name: raw_records[0][name] for name in _OMM_FIELDS} == _OMM_FIELDS + records = ommHandling.satOmm2elem(str(path)) + + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("namespace", ["urn:example:foreign", "urn:ccsds:schema:ndmxml:extension"]) +def test_omm_xml_foreign_namespace_has_no_records(tmp_path, namespace): + """The ndm prefix cannot make a foreign namespace into CCSDS OMM data.""" + path = _set_xml_namespace( + _writeXml(tmp_path / "foreign.xml", [_OMM_FIELDS]), "ndm", namespace + ) + + assert ommHandling._parseOmmXml(path.read_text(encoding="utf-8")) == [] + with pytest.raises(ValueError, match="found no OMM records"): + ommHandling.satOmm2elem(str(path)) + + def _write_omm_with_units(path, encoding, field_sets, unit_sets): """Add KVN annotations or XML attributes to selected numeric fields.""" if encoding == "kvn": @@ -134,9 +283,11 @@ def _write_omm_with_units(path, encoding, field_sets, unit_sets): return path -@pytest.mark.parametrize("encoding", ["kvn", "xml"]) +@pytest.mark.parametrize("encoding,namespace_prefix", [ + ("kvn", None), ("xml", None), ("xml", "ndm"), ("xml", ""), +]) @pytest.mark.parametrize("bstar_unit", ["1/ER", "1/[Earth radii]"]) -def test_omm_supported_units(tmp_path, capsys, encoding, bstar_unit): +def test_omm_supported_units(tmp_path, capsys, encoding, namespace_prefix, bstar_unit): """CCSDS unit declarations preserve the orbit and retained mean elements.""" units = { "MEAN_MOTION": "rev/day", @@ -149,6 +300,7 @@ def test_omm_supported_units(tmp_path, capsys, encoding, bstar_unit): "MEAN_MOTION_DDOT": "rev/day**3", } path = _write_omm_with_units(tmp_path / f"units.{encoding}", encoding, [_OMM_FIELDS], [units]) + _set_xml_namespace(path, namespace_prefix) reference_path = _WRITERS[encoding](tmp_path / f"reference.{encoding}", [_OMM_FIELDS]) records = ommHandling.satOmm2elem(str(path)) @@ -162,6 +314,23 @@ def test_omm_supported_units(tmp_path, capsys, encoding, bstar_unit): assert capsys.readouterr().out == "" +@pytest.mark.parametrize("namespace_prefix", ["ndm", ""]) +@pytest.mark.parametrize("field_name,value,units", [ + ("INCLINATION", _OMM_FIELDS["INCLINATION"], "rad"), + ("CENTER_NAME", "MARS", None), + ("MEAN_MOTION", "nan", None), + ("MEAN_MOTION", "0", None), # [rev/day] +]) +def test_omm_xml_namespace_invalid_record_is_skipped( + tmp_path, monkeypatch, capsys, namespace_prefix, field_name, value, units +): + """Qualified invalid values, metadata, and units are rejected before SGP4.""" + _check_invalid_record( + tmp_path, monkeypatch, capsys, "xml", field_name, value, + units=units, namespace_prefix=namespace_prefix, + ) + + @pytest.mark.parametrize("encoding", ["kvn", "xml"]) @pytest.mark.parametrize("field_name,unit", [ ("MEAN_MOTION", "rad/s"), @@ -650,13 +819,14 @@ def test_omm_bad_record_is_skipped_not_fatal(tmp_path): assert [d.satName for d in ommDataList] == ["ISS (ZARYA)"] +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) @pytest.mark.parametrize("omitted_blocks,missing_field", [ (("metadata",), "CENTER_NAME"), (("data",), "EPOCH"), (("metadata", "data"), "EPOCH"), ]) def test_omm_xml_incomplete_segments_preserve_record_numbers( - tmp_path, monkeypatch, capsys, omitted_blocks, missing_field + tmp_path, monkeypatch, capsys, namespace_prefix, omitted_blocks, missing_field ): """Incomplete XML segments are reported without shifting later diagnostics.""" broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") @@ -668,6 +838,7 @@ def test_omm_xml_incomplete_segments_preserve_record_numbers( for block in omitted_blocks: first_segment.remove(first_segment.find(block)) tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) initialize = ommHandling.sgp4omm.initialize initialized_names = [] expected_names = [_OMM_FIELDS["OBJECT_NAME"], second["OBJECT_NAME"]] @@ -691,7 +862,8 @@ def initialize_valid_record(satellite, fields): assert "MEAN_MOTION" in warnings[1] -def test_omm_xml_all_incomplete_segments_are_reported(tmp_path, capsys): +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +def test_omm_xml_all_incomplete_segments_are_reported(tmp_path, capsys, namespace_prefix): """A file of incomplete segments returns no records and warns for each segment.""" path = _writeXml(tmp_path / "incomplete.xml", [_OMM_FIELDS] * 3) tree = ET.parse(path) @@ -700,6 +872,7 @@ def test_omm_xml_all_incomplete_segments_are_reported(tmp_path, capsys): for block in blocks: segment.remove(segment.find(block)) tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) assert ommHandling.satOmm2elem(str(path)) == [] @@ -710,9 +883,11 @@ def test_omm_xml_all_incomplete_segments_are_reported(tmp_path, capsys): assert "required" in warning -def test_omm_xml_without_segments_raises(tmp_path): +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +def test_omm_xml_without_segments_raises(tmp_path, namespace_prefix): """An XML document containing no segments remains a file-level failure.""" path = _writeXml(tmp_path / "empty.xml", []) + _set_xml_namespace(path, namespace_prefix) with pytest.raises(ValueError, match="found no OMM records"): ommHandling.satOmm2elem(str(path)) @@ -843,7 +1018,9 @@ def test_omm_out_of_range_record_is_skipped( _check_invalid_record(tmp_path, monkeypatch, capsys, encoding, field_name, value) -def _check_invalid_record(tmp_path, monkeypatch, capsys, encoding, field_name, value, units=None): +def _check_invalid_record( + tmp_path, monkeypatch, capsys, encoding, field_name, value, units=None, namespace_prefix=None +): """Check record recovery and verify that SGP4 only receives the valid record.""" broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") broken[field_name] = value @@ -852,6 +1029,7 @@ def _check_invalid_record(tmp_path, monkeypatch, capsys, encoding, field_name, v _WRITERS[encoding](path, [broken, _OMM_FIELDS]) else: _write_omm_with_units(path, encoding, [broken, _OMM_FIELDS], [{field_name: units}, {}]) + _set_xml_namespace(path, namespace_prefix) initialize = ommHandling.sgp4omm.initialize initialized_names = [] diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index b54a218d58d..4a135ef6a2f 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -92,6 +92,14 @@ "MEAN_ELEMENT_THEORY": "SGP4", } +#: XML fields consumed by this reader; comments and unknown extension fields may repeat. +_OMM_XML_FIELDS = frozenset( + (*_REQUIRED_OMM_FIELDS, *_SUPPORTED_OMM_METADATA, "OBJECT_NAME") +) + +#: Preserve duplicate XML elements until validation can reject the individual record. +_DUPLICATE_OMM_XML_ELEMENT = object() + #: Human-readable names for the supported classification codes. _CLASSIFICATION_NAMES = {"U": "Unclassified", "C": "Classified", "S": "Secret"} @@ -257,31 +265,48 @@ def _parseOmmXml(text: str) -> list: """ Parse the XML (CCSDS NDM) encoding. + Accept unqualified elements and elements in the ``urn:ccsds:schema:ndmxml`` + namespace, independently of their prefix or use of a default namespace. Retain the available fields of every segment, including those missing metadata or data, so per-record validation reports their original positions in the file. + Mark duplicate OMM fields and blocks for rejection during per-record validation. :param text: full text of the XML file - :return: list of raw field dictionaries retaining numeric unit declarations + :return: list of raw field dictionaries retaining numeric unit declarations and duplicate markers """ root = ET.fromstring(text) + # Normalize only CCSDS tags so foreign namespaces cannot override OMM fields. + for element in root.iter(): + element.tag = element.tag.removeprefix("{urn:ccsds:schema:ndmxml}") records = [] for segment in root.findall(".//segment"): - metadata = segment.find("metadata") - data = segment.find("data") + blocks = { + "metadata": segment.findall("metadata"), + "data": segment.findall("data"), + } + if len(blocks["data"]) == 1: + data = blocks["data"][0] + blocks["meanElements"] = data.findall("meanElements") + blocks["tleParameters"] = data.findall("tleParameters") + duplicates = { + name: _DUPLICATE_OMM_XML_ELEMENT for name, matches in blocks.items() if len(matches) > 1 + } + if duplicates: + records.append(duplicates) + continue fields = {} - blocks = [metadata] - if data is not None: - blocks.extend([data.find("meanElements"), data.find("tleParameters")]) - for block in blocks: - if block is None: - continue - for element in block: - key = element.tag.upper() - value = (element.text or "").strip() - if key in _OMM_FIELD_UNITS and "units" in element.attrib: - value = _OmmFieldWithUnits(value, element.attrib["units"]) - fields[key] = value + for name in ("metadata", "meanElements", "tleParameters"): + for block in blocks.get(name, []): + for element in block: + key = element.tag.upper() + if key in _OMM_XML_FIELDS and key in fields: + fields[key] = _DUPLICATE_OMM_XML_ELEMENT + continue + value = (element.text or "").strip() + if key in _OMM_FIELD_UNITS and "units" in element.attrib: + value = _OmmFieldWithUnits(value, element.attrib["units"]) + fields[key] = value records.append(fields) return records @@ -444,6 +469,8 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: normalized = {} for key, value in fields.items(): key = str(key).upper() + if value is _DUPLICATE_OMM_XML_ELEMENT: + raise ValueError(f"Duplicate OMM XML element: {key}.") if isinstance(value, _OmmFieldWithUnits): supported_units = _OMM_FIELD_UNITS[key] if value.units not in supported_units: @@ -596,6 +623,10 @@ def satOmm2elem(omm_path: str) -> list: including nine-digit OMM IDs, are preserved in ``noradID`` without being constrained by SGP4's internal identifier storage. + XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, + using either prefixes or a default namespace. XML records containing duplicate + OMM fields or blocks are skipped with a warning before SGP4 is invoked. + Calendar-date and day-of-year epochs accept arbitrary fractional-second digits. Digits beyond microsecond precision are truncated consistently for ``ommEpoch``, SGP4 initialization and propagation, and frame conversion. From 1b1d73c883ed25ba3adeea0034acf38129834b57 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:24:41 -0700 Subject: [PATCH 02/15] [#1673] Accept CelesTrak SGP/SGP4 OMM theory label - Normalize the exact alias to SGP4 across all four encodings. - Preserve existing physical metadata and ephemeris-type validation. - Expand regression coverage and update the original OMM beta release note. Validation: 990 tests passed, plus pre-commit and focused Sphinx checks. --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 38 ++++++++++++++----- src/utilities/ommHandling.py | 9 ++++- 3 files changed, 37 insertions(+), 12 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index ae22fa8dff0..4d450df0933 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index c20f450b6cb..0b327aaca45 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -382,6 +382,7 @@ def test_omm_kvn_preserves_bracketed_names(tmp_path, capsys, name): @pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("mean_element_theory", ["SGP4", "SGP/SGP4"]) @pytest.mark.parametrize("field_name,value", [ ("CENTER_NAME", "MARS"), ("REF_FRAME", "GCRF"), @@ -391,6 +392,9 @@ def test_omm_kvn_preserves_bracketed_names(tmp_path, capsys, name): ("MEAN_ELEMENT_THEORY", "DSST"), ("MEAN_ELEMENT_THEORY", "SGP4-XP"), ("MEAN_ELEMENT_THEORY", "SGP"), + ("MEAN_ELEMENT_THEORY", "SGP/SGP4-XP"), + ("MEAN_ELEMENT_THEORY", "SGP4/SGP"), + ("MEAN_ELEMENT_THEORY", "SGP / SGP4"), ("EPHEMERIS_TYPE", "1"), ("EPHEMERIS_TYPE", "3"), ("EPHEMERIS_TYPE", "4"), @@ -399,10 +403,13 @@ def test_omm_kvn_preserves_bracketed_names(tmp_path, capsys, name): ("EPHEMERIS_TYPE", "99"), ]) def test_omm_unsupported_physical_metadata_is_skipped( - tmp_path, monkeypatch, capsys, encoding, field_name, value + tmp_path, monkeypatch, capsys, encoding, mean_element_theory, field_name, value ): - """Unsupported physical declarations never reach SGP4; valid records still return.""" - _check_invalid_record(tmp_path, monkeypatch, capsys, encoding, field_name, value) + """Neither accepted theory label relaxes validation; subsequent valid records survive.""" + _check_invalid_record( + tmp_path, monkeypatch, capsys, encoding, field_name, value, + mean_element_theory=mean_element_theory, + ) @pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @@ -462,18 +469,25 @@ def unexpected_initialize(*args, **kwargs): @pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @pytest.mark.parametrize("ephemeris_type", [None, "0", "2"]) -def test_omm_supported_physical_metadata(tmp_path, capsys, encoding, ephemeris_type): - """Supported metadata accepts case/whitespace variations and compatible ephemeris types.""" +@pytest.mark.parametrize("mean_element_theory", ["SGP4", " sgp4 ", "SGP/SGP4", " sgp/sgp4 "]) +def test_omm_supported_physical_metadata( + tmp_path, capsys, encoding, ephemeris_type, mean_element_theory +): + """CelesTrak theory aliases and case/whitespace variations preserve SGP4 orbital results.""" fields = dict(_OMM_FIELDS) for name in _PHYSICAL_METADATA_FIELDS: fields[name] = f" {_OMM_FIELDS[name].lower()} " + fields["MEAN_ELEMENT_THEORY"] = mean_element_theory if ephemeris_type is None: del fields["EPHEMERIS_TYPE"] else: fields["EPHEMERIS_TYPE"] = ephemeris_type path = _WRITERS[encoding](tmp_path / f"supported-metadata.{encoding}", [fields]) + reference_fields = dict(_OMM_FIELDS, EPHEMERIS_TYPE=ephemeris_type or "0") + reference_path = _WRITERS[encoding](tmp_path / f"reference.{encoding}", [reference_fields]) records = ommHandling.satOmm2elem(str(path)) + reference = ommHandling.satOmm2elem(str(reference_path))[0] assert len(records) == 1 record = records[0] @@ -481,7 +495,9 @@ def test_omm_supported_physical_metadata(tmp_path, capsys, encoding, ephemeris_t "EARTH", "TEME", "UTC", "SGP4", ) assert record.propagator == (ephemeris_type or "0") - assert all(np.isfinite(getattr(record.oe, name)) for name in ("a", "e", "i", "Omega", "omega", "f")) + for name in ("a", "e", "i", "Omega", "omega", "f"): + assert np.isfinite(getattr(record.oe, name)) + assert getattr(record.oe, name) == getattr(reference.oe, name) assert capsys.readouterr().out == "" @@ -1019,16 +1035,18 @@ def test_omm_out_of_range_record_is_skipped( def _check_invalid_record( - tmp_path, monkeypatch, capsys, encoding, field_name, value, units=None, namespace_prefix=None + tmp_path, monkeypatch, capsys, encoding, field_name, value, units=None, namespace_prefix=None, + mean_element_theory="SGP4", ): """Check record recovery and verify that SGP4 only receives the valid record.""" - broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") + valid = dict(_OMM_FIELDS, MEAN_ELEMENT_THEORY=mean_element_theory) + broken = dict(valid, OBJECT_NAME="BROKEN-SAT") broken[field_name] = value path = tmp_path / f"invalid-record.{encoding}" if units is None: - _WRITERS[encoding](path, [broken, _OMM_FIELDS]) + _WRITERS[encoding](path, [broken, valid]) else: - _write_omm_with_units(path, encoding, [broken, _OMM_FIELDS], [{field_name: units}, {}]) + _write_omm_with_units(path, encoding, [broken, valid], [{field_name: units}, {}]) _set_xml_namespace(path, namespace_prefix) initialize = ommHandling.sgp4omm.initialize initialized_names = [] diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index 4a135ef6a2f..11d428c6d61 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -429,7 +429,8 @@ def _normalize_omm_physical_metadata(fields: dict, omm_format: str) -> None: CelesTrak JSON and CSV may omit the four fixed metadata fields. XML and KVN must declare them. Explicitly empty or incompatible declarations are rejected in every encoding. Ephemeris type 0 is accepted for CelesTrak compatibility; - type 2 explicitly identifies SGP4. + type 2 explicitly identifies SGP4. CelesTrak's ``SGP/SGP4`` mean-element theory + label is normalized to ``SGP4`` in every encoding. :param fields: normalized field strings with numeric fields already validated :param omm_format: detected encoding (``xml``, ``kvn``, ``json``, or ``csv``) @@ -441,6 +442,8 @@ def _normalize_omm_physical_metadata(fields: dict, omm_format: str) -> None: raise ValueError(f"OMM {name} is required for {omm_format.upper()} records.") fields[name] = expected value = fields[name].strip().upper() + if name == "MEAN_ELEMENT_THEORY" and value == "SGP/SGP4": + value = "SGP4" if value != expected: raise ValueError(f"Unsupported OMM {name}: {fields[name]!r}; expected {expected}.") fields[name] = value @@ -638,6 +641,10 @@ def satOmm2elem(omm_path: str) -> list: fields explicitly. Empty or incompatible declarations cause the record to be skipped with a warning before SGP4 is invoked. + CelesTrak's ``MEAN_ELEMENT_THEORY = SGP/SGP4`` is accepted as an alias for + ``SGP4`` in every encoding, after trimming whitespace and normalizing case. + ``OmmData.meanElementTheory`` stores the canonical value ``SGP4``. + Numeric XML/KVN unit declarations must match the CCSDS units used by SGP4: ``deg``, ``rev/day``, ``rev/day**2``, ``rev/day**3``, and inverse Earth radii (``1/ER`` or ``1/[Earth radii]``) for ``BSTAR``. Dimensionless fields must From 0f6662a852d64ea407c494d1cba8a4b1b3ab07ab Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:38:22 -0700 Subject: [PATCH 03/15] [#1314] Reject duplicate fields across OMM encodings --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 82 +++++++++++++++++++ src/utilities/ommHandling.py | 54 ++++++++---- 3 files changed, 121 insertions(+), 17 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index 4d450df0933..c6a3347c0b7 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Rejects ambiguous duplicate consumed fields across all four encodings. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 0b327aaca45..203a0450199 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -23,7 +23,9 @@ # import copy +import csv import datetime as dt +import io import json import xml.etree.ElementTree as ET @@ -131,6 +133,86 @@ def _set_xml_namespace(path, prefix, namespace="urn:ccsds:schema:ndmxml"): _WRITERS = {"kvn": _writeKvn, "json": _writeJson, "csv": _writeCsv, "xml": _writeXml} +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv"]) +@pytest.mark.parametrize("field_name,bad_value", [ + ("CENTER_NAME", "MARS"), ("MEAN_MOTION", "nan"), ("OBJECT_NAME", "AMBIGUOUS"), +]) +@pytest.mark.parametrize("bad_first", [False, True]) +@pytest.mark.parametrize("mixed_case", [False, True]) +@pytest.mark.parametrize("third_occurrence", [False, True]) +def test_omm_duplicate_fields( + tmp_path, monkeypatch, capsys, encoding, field_name, bad_value, + bad_first, mixed_case, third_occurrence, +): + """Duplicates cannot overwrite invalid values, even after a third occurrence.""" + fields = dict(_OMM_FIELDS, ELEMENT_SET_NO="888") + pairs = [(name, value) for name, value in fields.items() if name != field_name] + values = [bad_value, fields[field_name]] if bad_first else [fields[field_name], bad_value] + pairs.extend(zip([field_name, field_name.lower() if mixed_case else field_name], values)) + if third_occurrence: + pairs.append((field_name, fields[field_name])) + path = tmp_path / f"duplicates.{encoding}" + if encoding == "json": + record = "{" + ",".join(f"{json.dumps(key)}:{json.dumps(value)}" for key, value in pairs) + "}" + text = f"[{record},{json.dumps(_OMM_FIELDS)}]" + elif encoding == "kvn": + text = "" + for record_pairs in (pairs, list(_OMM_FIELDS.items())): + text += "CCSDS_OMM_VERS = 2.0\n" + text += "".join(f"{key} = {value}\n" for key, value in record_pairs) + else: + stream = io.StringIO() + writer = csv.writer(stream) + writer.writerow([key for key, value in pairs]) + writer.writerow([value for key, value in pairs]) + text = stream.getvalue() + path.write_text(text, encoding="utf-8") + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, fields): + """Only the unambiguous trailing record may reach SGP4.""" + assert encoding != "csv" + assert fields["ELEMENT_SET_NO"] == _OMM_FIELDS["ELEMENT_SET_NO"] + initialized.append(fields["OBJECT_NAME"]) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + if encoding == "csv": + with pytest.raises(ValueError, match=f"Duplicate OMM CSV.*{field_name}"): + ommHandling.satOmm2elem(path) + assert initialized == [] + else: + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] + assert initialized == [_OMM_FIELDS["OBJECT_NAME"]] + warning = capsys.readouterr().out + assert "skipped OMM record 1" in warning + assert f"Duplicate OMM {encoding.upper()} field: {field_name}" in warning + + +@pytest.mark.parametrize("bad_first", [False, True]) +def test_omm_kvn_duplicate_units(tmp_path, capsys, bad_first): + """A duplicate numeric field cannot discard an unsupported unit declaration.""" + path = _writeKvn(tmp_path / "duplicate-units.kvn", [_OMM_FIELDS, _OMM_FIELDS]) + values = ["1.0 [rad]", "51.6416 [deg]"] # [rad], [deg] + if not bad_first: + values.reverse() + line = f"INCLINATION = {_OMM_FIELDS['INCLINATION']}" + path.write_text(path.read_text().replace(line, "\n".join(f"INCLINATION = {v}" for v in values), 1)) + assert len(ommHandling.satOmm2elem(path)) == 1 + assert "Duplicate OMM KVN field: INCLINATION" in capsys.readouterr().out + + +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv"]) +def test_omm_repeated_unknown_fields(tmp_path, capsys, encoding): + """Repeated extension fields remain harmless in each flat encoding.""" + fields = dict(_OMM_FIELDS, USER_DEFINED_NOTE="first", user_defined_note="second") + path = _WRITERS[encoding](tmp_path / f"extensions.{encoding}", [fields]) + assert len(ommHandling.satOmm2elem(path)) == 1 + assert capsys.readouterr().out == "" + + @pytest.mark.parametrize("namespace_prefix", ["ndm", "orbit", ""]) def test_omm_xml_namespace_preserves_fields_and_orbit(tmp_path, capsys, namespace_prefix): """CCSDS namespaces preserve all 21 fields, output metadata, and osculating elements.""" diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index 11d428c6d61..30654e5eecd 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -92,13 +92,13 @@ "MEAN_ELEMENT_THEORY": "SGP4", } -#: XML fields consumed by this reader; comments and unknown extension fields may repeat. -_OMM_XML_FIELDS = frozenset( +#: Fields consumed by this reader; comments and unknown extension fields may repeat. +_OMM_FIELDS = frozenset( (*_REQUIRED_OMM_FIELDS, *_SUPPORTED_OMM_METADATA, "OBJECT_NAME") ) -#: Preserve duplicate XML elements until validation can reject the individual record. -_DUPLICATE_OMM_XML_ELEMENT = object() +#: Preserve duplicate fields or XML blocks until validation can reject the record. +_DUPLICATE_OMM_FIELD = object() #: Human-readable names for the supported classification codes. _CLASSIFICATION_NAMES = {"U": "Unclassified", "C": "Classified", "S": "Secret"} @@ -205,6 +205,21 @@ def _detectOmmFormat(text: str) -> str: # ---------------------------------------------------------------------------------------------------------- # # OMM parsers # # ---------------------------------------------------------------------------------------------------------- # +def _collect_omm_fields(pairs) -> dict: + """Normalize field names while retaining duplicate consumed fields for rejection. + + :param pairs: iterable of field name/value pairs, including possible duplicates + :return: field dictionary with duplicate markers that survive later occurrences + """ + fields = {} + for key, value in pairs: + key = str(key).upper() + if key in _OMM_FIELDS and key in fields: + value = _DUPLICATE_OMM_FIELD + fields[key] = value + return fields + + def _parseOmmKvn(text: str) -> list: """ Parse the CCSDS keyword-value notation (KVN) encoding. @@ -235,7 +250,7 @@ def _parseOmmKvn(text: str) -> list: if key == "CCSDS_OMM_VERS" and current: records.append(current) current = {} - current[key] = value + current[key] = _DUPLICATE_OMM_FIELD if key in _OMM_FIELDS and key in current else value if current: records.append(current) @@ -253,7 +268,7 @@ def _parseOmmJson(text: str) -> list: :return: list of raw JSON records, to be validated and normalized individually :raises ValueError: if the JSON document is invalid or is not an object or array """ - payload = json.loads(text) + payload = json.loads(text, object_pairs_hook=_collect_omm_fields) if isinstance(payload, dict): return [payload] if not isinstance(payload, list): @@ -289,7 +304,7 @@ def _parseOmmXml(text: str) -> list: blocks["meanElements"] = data.findall("meanElements") blocks["tleParameters"] = data.findall("tleParameters") duplicates = { - name: _DUPLICATE_OMM_XML_ELEMENT for name, matches in blocks.items() if len(matches) > 1 + name: _DUPLICATE_OMM_FIELD for name, matches in blocks.items() if len(matches) > 1 } if duplicates: records.append(duplicates) @@ -300,8 +315,8 @@ def _parseOmmXml(text: str) -> list: for block in blocks.get(name, []): for element in block: key = element.tag.upper() - if key in _OMM_XML_FIELDS and key in fields: - fields[key] = _DUPLICATE_OMM_XML_ELEMENT + if key in _OMM_FIELDS and key in fields: + fields[key] = _DUPLICATE_OMM_FIELD continue value = (element.text or "").strip() if key in _OMM_FIELD_UNITS and "units" in element.attrib: @@ -320,10 +335,14 @@ def _parseOmmCsv(text: str) -> list: """ import io + reader = sgp4omm.parse_csv(io.StringIO(text)) + headers = _collect_omm_fields((name, "") for name in reader.fieldnames or []) + duplicates = [name for name, value in headers.items() if value is _DUPLICATE_OMM_FIELD] + if duplicates: + raise ValueError(f"Duplicate OMM CSV field(s) in header: {', '.join(duplicates)}.") records = [] - for row in sgp4omm.parse_csv(io.StringIO(text)): - records.append({str(key).upper(): ("" if value is None else str(value)) - for key, value in row.items()}) + for row in reader: + records.append(_collect_omm_fields(row.items())) return records @@ -470,10 +489,10 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: # Check declared units before discarding them or supplying optional defaults. # Validation stays inside the per-record error handler for XML and KVN alike. normalized = {} - for key, value in fields.items(): - key = str(key).upper() - if value is _DUPLICATE_OMM_XML_ELEMENT: - raise ValueError(f"Duplicate OMM XML element: {key}.") + for key, value in _collect_omm_fields(fields.items()).items(): + if value is _DUPLICATE_OMM_FIELD: + kind = "element" if omm_format == "xml" else "field" + raise ValueError(f"Duplicate OMM {omm_format.upper()} {kind}: {key}.") if isinstance(value, _OmmFieldWithUnits): supported_units = _OMM_FIELD_UNITS[key] if value.units not in supported_units: @@ -629,6 +648,9 @@ def satOmm2elem(omm_path: str) -> list: XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, using either prefixes or a default namespace. XML records containing duplicate OMM fields or blocks are skipped with a warning before SGP4 is invoked. + KVN and JSON records with duplicate consumed fields are also skipped, including + names that collide after case normalization. Duplicate consumed CSV headers + make the whole file ambiguous and raise ``ValueError`` before conversion. Calendar-date and day-of-year epochs accept arbitrary fractional-second digits. Digits beyond microsecond precision are truncated consistently for ``ommEpoch``, From 28a8e18678ea6dafcc87da0a7ad3850466f1b73d Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:40:21 -0700 Subject: [PATCH 04/15] [#1314] Validate OMM CSV rows and preserve quoted fields --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 75 ++++++++++++++++++- src/utilities/ommHandling.py | 39 ++++++++-- 3 files changed, 104 insertions(+), 12 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index c6a3347c0b7..a7f5c9ca246 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Rejects ambiguous duplicate consumed fields across all four encodings. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 203a0450199..631b5e5685a 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -85,9 +85,10 @@ def _writeJson(path, fieldSets): def _writeCsv(path, fieldSets): fieldSets = list(fieldSets) - header = ",".join(fieldSets[0].keys()) - rows = [",".join(str(v) for v in fields.values()) for fields in fieldSets] - path.write_text("\n".join([header] + rows) + "\n", encoding="utf-8") + with path.open("w", encoding="utf-8", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=list(fieldSets[0])) + writer.writeheader() + writer.writerows(fieldSets) return path @@ -213,6 +214,74 @@ def test_omm_repeated_unknown_fields(tmp_path, capsys, encoding): assert capsys.readouterr().out == "" +@pytest.mark.parametrize("cell_delta", [-3, -1, 1, 3]) +def test_omm_csv_row_width(tmp_path, monkeypatch, capsys, cell_delta): + """Damaged rows cannot turn missing cells into defaults or lose extra cells.""" + fields = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") + row = list(fields.values()) + row = row[:cell_delta] if cell_delta < 0 else row + ["extra"] * cell_delta + incomplete = dict(_OMM_FIELDS, MEAN_MOTION="") + stream = io.StringIO() + writer = csv.writer(stream) + writer.writerow(list(fields)) + writer.writerows([row, list(_OMM_FIELDS.values()), list(incomplete.values()), list(_OMM_FIELDS.values())]) + path = tmp_path / "row-width.csv" + path.write_text(stream.getvalue(), encoding="utf-8") + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, fields): + """Only complete records may reach the backend.""" + assert fields["OBJECT_NAME"] == _OMM_FIELDS["OBJECT_NAME"] + assert fields["MEAN_MOTION"] == _OMM_FIELDS["MEAN_MOTION"] + initialized.append(fields) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert len(records) == len(initialized) == 2 + assert all(record.bStar == float(_OMM_FIELDS["BSTAR"]) for record in records) + warnings = capsys.readouterr().out.splitlines() + assert len(warnings) == 2 + assert "skipped OMM record 1" in warnings[0] + assert f"{len(row)} cells; expected {len(fields)}" in warnings[0] + assert "skipped OMM record 3" in warnings[1] + assert "MEAN_MOTION" in warnings[1] + + +@pytest.mark.parametrize("prefix", ["", "\n", "\r\n\r\n", " \t\r\n\n"]) +def test_omm_csv_blank_lines_and_quoted_fields(tmp_path, capsys, prefix): + """Leading blank lines do not affect quoted Unicode and multiline field values.""" + name = 'Satellite, "quoted"\nsecond line — café' + fields = dict(_OMM_FIELDS, OBJECT_NAME=name) + path = _writeCsv(tmp_path / "quoted.csv", [fields, _OMM_FIELDS]) + path.write_text(prefix + path.read_text(encoding="utf-8"), encoding="utf-8") + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == [name, _OMM_FIELDS["OBJECT_NAME"]] + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("omit_column", [False, True]) +def test_omm_csv_optional_empty_or_absent_column(tmp_path, capsys, omit_column): + """An absent optional column or explicit empty cell still receives its default.""" + fields = dict(_OMM_FIELDS, BSTAR="") + if omit_column: + del fields["BSTAR"] + path = _writeCsv(tmp_path / "optional.csv", [fields]) + records = ommHandling.satOmm2elem(path) + assert len(records) == 1 + assert records[0].bStar == 0.0 # [1/Earth radii] + assert capsys.readouterr().out == "" + + +def test_omm_csv_unterminated_quote(tmp_path): + """Unrecoverable CSV quoting errors remain file-level failures.""" + path = _writeCsv(tmp_path / "truncated.csv", [_OMM_FIELDS]) + path.write_text(path.read_text() + '"unterminated\n', encoding="utf-8") + with pytest.raises(csv.Error): + ommHandling.satOmm2elem(path) + + @pytest.mark.parametrize("namespace_prefix", ["ndm", "orbit", ""]) def test_omm_xml_namespace_preserves_fields_and_orbit(tmp_path, capsys, namespace_prefix): """CCSDS namespaces preserve all 21 fields, output metadata, and osculating elements.""" diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index 30654e5eecd..7df1bf2928a 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -38,7 +38,9 @@ Earth-centered, TEME, UTC, SGP4 profile is supported. """ +import csv import datetime as dt +import io import json import xml.etree.ElementTree as ET from dataclasses import dataclass, field @@ -130,6 +132,13 @@ class _OmmFieldWithUnits: units: str +@dataclass(frozen=True) +class _OmmInvalidRecord: + """Retain a parser diagnostic at the original record position.""" + + reason: str + + # ---------------------------------------------------------------------------------------------------------- # # OMM Data Class # # ---------------------------------------------------------------------------------------------------------- # @@ -328,21 +337,30 @@ def _parseOmmXml(text: str) -> list: def _parseOmmCsv(text: str) -> list: """ - Parse the CSV encoding via the ``sgp4`` reader so the field naming matches upstream. + Parse CSV records with explicit header and row-width validation. :param text: full text of the CSV file - :return: list of dictionaries of raw OMM field strings + :return: field dictionaries or invalid-record markers, in original record order + :raises ValueError: if consumed header names are duplicated + :raises csv.Error: if CSV quoting is malformed """ - import io - - reader = sgp4omm.parse_csv(io.StringIO(text)) - headers = _collect_omm_fields((name, "") for name in reader.fieldnames or []) - duplicates = [name for name, value in headers.items() if value is _DUPLICATE_OMM_FIELD] + # Match format detection's leading-whitespace handling without changing quoted cells. + reader = csv.reader(io.StringIO(text.lstrip(), newline=""), strict=True) + headers = next(reader, []) + fields = _collect_omm_fields((name, "") for name in headers) + duplicates = [name for name, value in fields.items() if value is _DUPLICATE_OMM_FIELD] if duplicates: raise ValueError(f"Duplicate OMM CSV field(s) in header: {', '.join(duplicates)}.") records = [] for row in reader: - records.append(_collect_omm_fields(row.items())) + if not row: + continue + if len(row) != len(headers): + records.append(_OmmInvalidRecord( + f"OMM CSV row has {len(row)} cells; expected {len(headers)} from the header." + )) + continue + records.append(_collect_omm_fields(zip(headers, row))) return records @@ -483,6 +501,8 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: :return: a normalized copy safe to hand to SGP4 :raises ValueError: if the record is not a field dictionary or its fields are invalid """ + if isinstance(fields, _OmmInvalidRecord): + raise ValueError(fields.reason) if not isinstance(fields, dict): raise ValueError(f"OMM record must be a field object, got {type(fields).__name__}.") @@ -651,6 +671,9 @@ def satOmm2elem(omm_path: str) -> list: KVN and JSON records with duplicate consumed fields are also skipped, including names that collide after case normalization. Duplicate consumed CSV headers make the whole file ambiguous and raise ``ValueError`` before conversion. + CSV accepts leading blank lines and quoted cells, including embedded newlines. + Rows with missing or extra cells are skipped before optional defaults are applied; + optional columns omitted from the header still receive their normal defaults. Calendar-date and day-of-year epochs accept arbitrary fractional-second digits. Digits beyond microsecond precision are truncated consistently for ``ommEpoch``, From da164ac692a803a2c47f2c4fc6f895920c50de1f Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:42:08 -0700 Subject: [PATCH 05/15] [#1314] Validate OMM XML ancestry and field placement --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 125 ++++++++++++++++++ src/utilities/ommHandling.py | 120 ++++++++++++----- 3 files changed, 214 insertions(+), 33 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index a7f5c9ca246..3525c7fac02 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 631b5e5685a..d576b324ba0 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -416,6 +416,131 @@ def test_omm_xml_foreign_namespace_has_no_records(tmp_path, namespace): ommHandling.satOmm2elem(str(path)) +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +@pytest.mark.parametrize("standalone", [False, True]) +def test_omm_xml_message_containers(tmp_path, capsys, namespace_prefix, standalone): + """Standalone OMMs and direct OMM siblings in NDM containers retain record order.""" + second = dict(_OMM_FIELDS, OBJECT_NAME="SECOND-SAT", NORAD_CAT_ID="25545") + path = _writeXml(tmp_path / "containers.xml", [_OMM_FIELDS, second]) + tree = ET.parse(path) + if standalone: + tree._setroot(tree.find("omm")) + else: + message = tree.find("omm") + other = copy.deepcopy(message) + body = message.find("body") + body.remove(body.findall("segment")[1]) + other.find("body").remove(other.find("body/segment")) + tree.getroot().append(other) + # A different message family must not produce OMM records or warnings. + foreign_message = copy.deepcopy(message) + foreign_message.tag = "oem" + tree.getroot().insert(0, foreign_message) + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"], "SECOND-SAT"] + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +@pytest.mark.parametrize("parent_path", [".", "omm", "omm/body", "omm/body/segment/data"]) +def test_omm_xml_extensions_cannot_inject_segments(tmp_path, capsys, namespace_prefix, parent_path): + """Recognized descendants inside foreign extensions never become extra records.""" + path = _writeXml(tmp_path / "cached.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + cached_message = copy.deepcopy(tree.find("omm")) + cached_segment = copy.deepcopy(tree.find(".//segment")) + extension = ET.SubElement(tree.find(parent_path), "{urn:foreign}cache") + extension.extend([cached_message, cached_segment]) + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("root_name", ["opm", "wrapper", "{urn:foreign}omm"]) +def test_omm_xml_unrecognized_root(tmp_path, root_name): + """An arbitrary wrapper cannot turn its descendants into an OMM document.""" + path = _writeXml(tmp_path / "wrong-root.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + tree.getroot().tag = root_name + tree.write(path, encoding="utf-8") + with pytest.raises(ValueError, match="found no OMM records"): + ommHandling.satOmm2elem(path) + + +@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) +@pytest.mark.parametrize("field_name", list(_OMM_FIELDS)) +@pytest.mark.parametrize("target_offset", [1, 2]) +def test_omm_xml_field_block_placement( + tmp_path, monkeypatch, capsys, namespace_prefix, field_name, target_offset, +): + """Every consumed field is rejected in either of the other two field blocks.""" + broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") + incomplete = {name: value for name, value in broken.items() if name != "MEAN_MOTION"} + path = _writeXml(tmp_path / "misplaced.xml", [broken, _OMM_FIELDS, incomplete]) + tree = ET.parse(path) + segment = tree.find(".//segment") + parent = segment.find(f".//{field_name}/..") + element = parent.find(field_name) + block_names = ["metadata", "meanElements", "tleParameters"] + target = block_names[(block_names.index(parent.tag) + target_offset) % len(block_names)] + parent.remove(element) + segment.find(f".//{target}").append(element) + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, fields): + """Malformed XML must be rejected before propagation.""" + assert fields["OBJECT_NAME"] == _OMM_FIELDS["OBJECT_NAME"] + initialized.append(fields) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert len(records) == len(initialized) == 1 + warnings = capsys.readouterr().out.splitlines() + assert len(warnings) == 2 + assert "skipped OMM record 1" in warnings[0] + assert f"Misplaced OMM XML element {field_name} in {target}" in warnings[0] + assert "skipped OMM record 3" in warnings[1] + + +@pytest.mark.parametrize("case", ["metadata-fields", "misplaced-block", "direct-field", "nested-field", "body"]) +def test_omm_xml_malformed_structure(tmp_path, capsys, case): + """Block relocation, scalar children, and duplicate bodies remain record errors.""" + path = _writeXml(tmp_path / "structure.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + message = tree.find("omm") + valid_message = copy.deepcopy(message) + segment = message.find("body/segment") + if case == "metadata-fields": + metadata = segment.find("metadata") + segment.find("data/meanElements").extend(list(metadata)) + segment.remove(metadata) + elif case == "misplaced-block": + mean = segment.find("data/meanElements") + segment.find("data").remove(mean) + segment.append(mean) + elif case == "direct-field": + ET.SubElement(segment.find("data"), "BSTAR").text = "0.0" # [1/Earth radii] + elif case == "nested-field": + ET.SubElement(segment.find(".//INCLINATION"), "extra").text = "invalid" + else: + ET.SubElement(message, "body") + tree.getroot().append(valid_message) + tree.write(path, encoding="utf-8") + assert len(ommHandling.satOmm2elem(path)) == 1 + warnings = capsys.readouterr().out.splitlines() + assert len(warnings) == 1 + assert "skipped OMM record 1" in warnings[0] + assert "XML" in warnings[0] + + def _write_omm_with_units(path, encoding, field_sets, unit_sets): """Add KVN annotations or XML attributes to selected numeric fields.""" if encoding == "kvn": diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index 7df1bf2928a..6fa79b71ca9 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -94,6 +94,22 @@ "MEAN_ELEMENT_THEORY": "SGP4", } +#: Consumed XML fields must appear in their CCSDS owning blocks. +_OMM_XML_BLOCK_FIELDS = { + "metadata": frozenset(("OBJECT_NAME", "OBJECT_ID", *_SUPPORTED_OMM_METADATA)), + "meanElements": frozenset(( + "EPOCH", "MEAN_MOTION", "ECCENTRICITY", "INCLINATION", + "RA_OF_ASC_NODE", "ARG_OF_PERICENTER", "MEAN_ANOMALY", + )), + "tleParameters": frozenset(( + "EPHEMERIS_TYPE", "CLASSIFICATION_TYPE", "NORAD_CAT_ID", "ELEMENT_SET_NO", + "REV_AT_EPOCH", "BSTAR", "MEAN_MOTION_DOT", "MEAN_MOTION_DDOT", + )), +} + +#: Recognized structural tags, used to reject blocks in the wrong location. +_OMM_XML_BLOCKS = frozenset(("ndm", "omm", "body", "segment", "data", *_OMM_XML_BLOCK_FIELDS)) + #: Fields consumed by this reader; comments and unknown extension fields may repeat. _OMM_FIELDS = frozenset( (*_REQUIRED_OMM_FIELDS, *_SUPPORTED_OMM_METADATA, "OBJECT_NAME") @@ -285,6 +301,62 @@ def _parseOmmJson(text: str) -> list: return payload +def _validate_omm_xml_children(block, allowed) -> None: + """Check consumed fields and recognized blocks without interpreting extensions. + + :param block: XML element whose direct children are being validated + :param allowed: field or block names permitted at this location + :raises ValueError: if a consumed field or recognized block is misplaced + """ + for child in block: + name = child.tag.upper() if child.tag.upper() in _OMM_FIELDS else child.tag + if name in _OMM_FIELDS or name in _OMM_XML_BLOCKS: + if name not in allowed: + raise ValueError(f"Misplaced OMM XML element {name} in {block.tag}.") + if name in _OMM_FIELDS and len(child): + raise ValueError(f"OMM XML field {name} must contain text without child elements.") + + +def _parse_omm_xml_segment(segment) -> dict: + """Extract one segment while checking block placement and duplicate elements. + + :param segment: OMM segment with normalized CCSDS tag names + :return: raw fields, retaining units and duplicate markers + :raises ValueError: if a field or block is misplaced + """ + blocks = { + "metadata": segment.findall("metadata"), + "data": segment.findall("data"), + } + if len(blocks["data"]) == 1: + data = blocks["data"][0] + blocks["meanElements"] = data.findall("meanElements") + blocks["tleParameters"] = data.findall("tleParameters") + duplicates = { + name: _DUPLICATE_OMM_FIELD for name, matches in blocks.items() if len(matches) > 1 + } + if duplicates: + return duplicates + + _validate_omm_xml_children(segment, {"metadata", "data"}) + for data in blocks["data"]: + _validate_omm_xml_children(data, {"meanElements", "tleParameters"}) + fields = {} + for name, allowed in _OMM_XML_BLOCK_FIELDS.items(): + for block in blocks.get(name, []): + _validate_omm_xml_children(block, allowed) + for element in block: + key = element.tag.upper() + if key in _OMM_FIELDS and key in fields: + fields[key] = _DUPLICATE_OMM_FIELD + continue + value = (element.text or "").strip() + if key in _OMM_FIELD_UNITS and "units" in element.attrib: + value = _OmmFieldWithUnits(value, element.attrib["units"]) + fields[key] = value + return fields + + def _parseOmmXml(text: str) -> list: """ Parse the XML (CCSDS NDM) encoding. @@ -296,42 +368,22 @@ def _parseOmmXml(text: str) -> list: Mark duplicate OMM fields and blocks for rejection during per-record validation. :param text: full text of the XML file - :return: list of raw field dictionaries retaining numeric unit declarations and duplicate markers + :return: raw fields or invalid-record markers, retaining segment positions """ root = ET.fromstring(text) # Normalize only CCSDS tags so foreign namespaces cannot override OMM fields. for element in root.iter(): element.tag = element.tag.removeprefix("{urn:ccsds:schema:ndmxml}") + messages = [root] if root.tag == "omm" else root.findall("omm") if root.tag == "ndm" else [] records = [] - for segment in root.findall(".//segment"): - blocks = { - "metadata": segment.findall("metadata"), - "data": segment.findall("data"), - } - if len(blocks["data"]) == 1: - data = blocks["data"][0] - blocks["meanElements"] = data.findall("meanElements") - blocks["tleParameters"] = data.findall("tleParameters") - duplicates = { - name: _DUPLICATE_OMM_FIELD for name, matches in blocks.items() if len(matches) > 1 - } - if duplicates: - records.append(duplicates) - continue - - fields = {} - for name in ("metadata", "meanElements", "tleParameters"): - for block in blocks.get(name, []): - for element in block: - key = element.tag.upper() - if key in _OMM_FIELDS and key in fields: - fields[key] = _DUPLICATE_OMM_FIELD - continue - value = (element.text or "").strip() - if key in _OMM_FIELD_UNITS and "units" in element.attrib: - value = _OmmFieldWithUnits(value, element.attrib["units"]) - fields[key] = value - records.append(fields) + for message in messages: + for segment in message.findall("body/segment"): + try: + if len(message.findall("body")) != 1: + raise ValueError("Duplicate OMM XML element: BODY.") + records.append(_parse_omm_xml_segment(segment)) + except ValueError as error: + records.append(_OmmInvalidRecord(str(error))) return records @@ -666,8 +718,12 @@ def satOmm2elem(omm_path: str) -> list: SGP4's internal identifier storage. XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, - using either prefixes or a default namespace. XML records containing duplicate - OMM fields or blocks are skipped with a warning before SGP4 is invoked. + using either prefixes or a default namespace. Records are read from + ``omm/body/segment`` in a standalone OMM or a direct OMM child of an NDM container. + Segments inside extensions or other message types are ignored. Consumed fields + must appear in their owning metadata, mean-elements or TLE-parameters blocks. + XML records containing misplaced or duplicate OMM fields or blocks are skipped + with a warning before SGP4 is invoked. KVN and JSON records with duplicate consumed fields are also skipped, including names that collide after case normalization. Duplicate consumed CSV headers make the whole file ambiguous and raise ``ValueError`` before conversion. From bc4b0d064b512eca570dab86e829bd9ecd36e845 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:43:15 -0700 Subject: [PATCH 06/15] [#1314] Accept signed OMM catalog IDs within the nine-digit range --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 32 +++++++++++++++---- src/utilities/ommHandling.py | 9 ++++-- 3 files changed, 33 insertions(+), 10 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index 3525c7fac02..ed556a1c62d 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, lifting the five-digit catalog number limit of the TLE format. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index d576b324ba0..0ceaa388aaf 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -883,11 +883,13 @@ def test_omm_encodings_agree(tmp_path): @pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @pytest.mark.parametrize("catalog_id", [ "00001", "99999", "100000", "100001", "339999", "340000", - "999999", "799500001", "999999999", + "999999", "799500001", "999999999", "0", "+0", "+00001", "+25544", + "+999999999", "000999999999", ]) -def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id): +@pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) +def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id, satrec): """Catalog IDs survive ingestion unchanged without affecting the orbital elements.""" - monkeypatch.setattr(ommHandling, "Satrec", PythonSatrec) + monkeypatch.setattr(ommHandling, "Satrec", satrec) fields = dict(_OMM_FIELDS) fields["NORAD_CAT_ID"] = catalog_id @@ -905,15 +907,31 @@ def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id): assert getattr(actual.oe, name) == getattr(reference.oe, name), name -@pytest.mark.parametrize("catalog_id", ["A0001", "100000.5", "-1", "nan"]) -def test_omm_invalid_catalog_number_is_skipped(tmp_path, capsys, catalog_id): - """Substituting an internal SGP4 ID must not admit malformed catalog numbers.""" +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("catalog_id", [ + "A0001", "100000.5", "-1", "nan", "+", "++25544", "+ 25544", "+-1", + "1000000000", "+1000000000", "0001000000000", "١٢٣", "１２３", +]) +def test_omm_invalid_catalog_number_is_skipped(tmp_path, monkeypatch, capsys, encoding, catalog_id): + """Malformed or out-of-range catalog IDs never reach SGP4; later records survive.""" fields = dict(_OMM_FIELDS, NORAD_CAT_ID=catalog_id) - path = _writeJson(tmp_path / "invalid-catalog.json", [fields, _OMM_FIELDS]) + fields["OBJECT_NAME"] = "BROKEN-SAT" + path = _WRITERS[encoding](tmp_path / f"invalid-catalog.{encoding}", [fields, _OMM_FIELDS]) + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, fields): + """Only the valid catalog entry may reach SGP4.""" + assert fields["OBJECT_NAME"] == _OMM_FIELDS["OBJECT_NAME"] + initialized.append(fields) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) records = ommHandling.satOmm2elem(str(path)) assert [record.noradID for record in records] == [_OMM_FIELDS["NORAD_CAT_ID"]] + assert len(initialized) == 1 warning = capsys.readouterr().out assert "skipped OMM record 1" in warning assert "NORAD_CAT_ID" in warning diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index 6fa79b71ca9..a62fa5a4e26 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -590,10 +590,14 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: # Validate the external ID before replacing it in the private SGP4 input. catalog_id = normalized["NORAD_CAT_ID"].strip() - if not catalog_id.isascii() or not catalog_id.isdecimal(): + catalog_digits = catalog_id.removeprefix("+") + if not catalog_digits.isascii() or not catalog_digits.isdecimal(): raise ValueError( f"satOmm2elem() requires NORAD_CAT_ID to be a non-negative decimal integer: {catalog_id!r}" ) + # Check the value's range without converting arbitrarily long digit strings to int. + if len(catalog_digits.lstrip("0")) > 9: + raise ValueError(f"OMM NORAD_CAT_ID must be between 0 and 999999999: {catalog_id!r}") _validate_omm_numeric_fields(normalized) _normalize_omm_physical_metadata(normalized, omm_format) @@ -715,7 +719,8 @@ def satOmm2elem(omm_path: str) -> list: The encoding (XML, JSON, CSV or KVN) is detected from the file contents. Catalog numbers, including nine-digit OMM IDs, are preserved in ``noradID`` without being constrained by - SGP4's internal identifier storage. + SGP4's internal identifier storage. Values must be in ``[0, 999999999]``; + optional leading ``+`` signs and leading zeros are accepted and preserved. XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, using either prefixes or a default namespace. Records are read from From 3a4aef93b34ab16bcc3ed9137272038b55ce405e Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:44:35 -0700 Subject: [PATCH 07/15] [#1314] Preserve missing OMM designators in public metadata --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 49 +++++++++++++++++++ src/utilities/ommHandling.py | 24 ++++----- 3 files changed, 63 insertions(+), 12 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index ed556a1c62d..25ce33d7c0c 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 0ceaa388aaf..ed7f7d1a078 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -966,6 +966,55 @@ def test_omm_optional_fields_defaulted(tmp_path): assert ommDataList[0].revAtEpoch == 0 +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("designator", [None, "", " \t ", "UNKNOWN", "unknown", "X", "1998-067A"]) +@pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) +def test_omm_designator_metadata(tmp_path, monkeypatch, capsys, encoding, designator, satrec): + """Public designators remain faithful while private backend identifiers stay safe.""" + fields = dict(_OMM_FIELDS) + if designator is None: + del fields["OBJECT_ID"] + else: + fields["OBJECT_ID"] = designator + original_fields = dict(fields) + path = _WRITERS[encoding](tmp_path / f"designator.{encoding}", [fields]) + reference_path = _WRITERS[encoding](tmp_path / f"reference.{encoding}", [_OMM_FIELDS]) + monkeypatch.setattr(ommHandling, "Satrec", satrec) + initialize = ommHandling.sgp4omm.initialize + initialized_designators = [] + + def initialize_with_private_identifier(satellite, backend_fields): + """Capture the backend's input and resulting internal designator.""" + initialized_designators.append(backend_fields["OBJECT_ID"]) + result = initialize(satellite, backend_fields) + expected = "98067A" if backend_fields["OBJECT_ID"] else "" + assert satellite.intldesg == expected + return result + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_with_private_identifier) + expected = "" if designator is None else designator.strip() + normalized = ommHandling._normalizeOmmFields(fields, encoding) + assert fields == original_fields + assert normalized["OBJECT_ID"] == expected + records = ommHandling.satOmm2elem(path) + reference = ommHandling.satOmm2elem(reference_path)[0] + assert len(records) == 1 + assert records[0].objectID == expected + assert initialized_designators == [expected if expected == "1998-067A" else "", "1998-067A"] + for name in ("a", "e", "i", "Omega", "omega", "f"): + assert getattr(records[0].oe, name) == getattr(reference.oe, name) + assert capsys.readouterr().out == "" + + +def test_omm_json_null_designator(tmp_path, capsys): + """JSON null designators return empty metadata without fabricating an identifier.""" + path = _writeJson(tmp_path / "null-designator.json", [dict(_OMM_FIELDS, OBJECT_ID=None)]) + records = ommHandling.satOmm2elem(path) + assert len(records) == 1 + assert records[0].objectID == "" + assert capsys.readouterr().out == "" + + @pytest.mark.parametrize("epochStr,expectedSecond", [ ("2026-09-10T12:00:00.000000", 0), ("2026-09-10T12:00:30", 30), diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index a62fa5a4e26..fb744c74c31 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -54,7 +54,7 @@ # same frame through identical precession, nutation and frame-bias terms. from Basilisk.utilities.tleHandling import _teme2j2000 -#: Fields that :py:func:`sgp4.omm.initialize` dereferences unconditionally. +#: Fields required in normalized records before conversion. _REQUIRED_OMM_FIELDS = ( "EPOCH", "MEAN_MOTION", @@ -65,7 +65,6 @@ "MEAN_ANOMALY", "NORAD_CAT_ID", "CLASSIFICATION_TYPE", - "OBJECT_ID", "EPHEMERIS_TYPE", "ELEMENT_SET_NO", "REV_AT_EPOCH", @@ -77,7 +76,6 @@ #: Defaults for fields that CelesTrak sometimes omits but SGP4 still requires. _OPTIONAL_OMM_DEFAULTS = { "CLASSIFICATION_TYPE": "U", - "OBJECT_ID": "0000-000A", "EPHEMERIS_TYPE": "0", "ELEMENT_SET_NO": "999", "REV_AT_EPOCH": "0", @@ -111,9 +109,7 @@ _OMM_XML_BLOCKS = frozenset(("ndm", "omm", "body", "segment", "data", *_OMM_XML_BLOCK_FIELDS)) #: Fields consumed by this reader; comments and unknown extension fields may repeat. -_OMM_FIELDS = frozenset( - (*_REQUIRED_OMM_FIELDS, *_SUPPORTED_OMM_METADATA, "OBJECT_NAME") -) +_OMM_FIELDS = frozenset().union(*_OMM_XML_BLOCK_FIELDS.values()) #: Preserve duplicate fields or XML blocks until validation can reject the record. _DUPLICATE_OMM_FIELD = object() @@ -174,7 +170,7 @@ class OmmData: # Optional metadata satName: str = field(default="BSK-Sat-00") # OBJECT_NAME noradID: str = field(default="00000") # NORAD_CAT_ID; unlike a TLE this may be >= 100000 - objectID: str = field(default="0000-000A") # international designator, e.g. "1998-067A" + objectID: str = field(default="") # international designator; empty when unavailable classification: str = field(default="Unknown") # "Unclassified", "Classified", "Secret" revAtEpoch: int = field(default=0) # revolution number at epoch [-] propagator: str = field(default="0") # EPHEMERIS_TYPE: 0 (CelesTrak default) or 2 (SGP4) @@ -614,9 +610,8 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: epoch = _parseOmmEpoch(normalized["EPOCH"]) normalized["EPOCH"] = epoch.strftime("%Y-%m-%dT%H:%M:%S.%f") - # The OBJECT_ID slice taken by sgp4 assumes the "YYYY-NNNP" international designator. - if len(normalized["OBJECT_ID"]) < 3: - normalized["OBJECT_ID"] = _OPTIONAL_OMM_DEFAULTS["OBJECT_ID"] + # Preserve missing or unknown designators in public metadata without inventing an ID. + normalized["OBJECT_ID"] = normalized.get("OBJECT_ID", "").strip() return normalized @@ -636,7 +631,7 @@ def _ommFields2Data(fields: dict, epoch: dt.datetime) -> OmmData: ommEpoch=epoch, satName=fields.get("OBJECT_NAME", "BSK-Sat-00").strip() or "BSK-Sat-00", noradID=str(fields["NORAD_CAT_ID"]).strip(), - objectID=fields.get("OBJECT_ID", "0000-000A").strip(), + objectID=fields.get("OBJECT_ID", "").strip(), classification=_CLASSIFICATION_NAMES[fields["CLASSIFICATION_TYPE"]], revAtEpoch=int(fields["REV_AT_EPOCH"]), propagator=str(fields.get("EPHEMERIS_TYPE", "0")).strip(), @@ -674,6 +669,10 @@ def _convertOmmMean2osculating(fields: dict, ommData: OmmData) -> om.ClassicElem # SGP4 limits its internal Alpha-5 ID to 339999, but the ID does not affect # propagation. Preserve the real catalog number in fields and ommData. sgp4_fields = dict(fields, NORAD_CAT_ID="0") + # Older SGP4 initializers slice every OBJECT_ID as a YYYY-NNNP designator. + # Use an empty internal identifier for unavailable or non-designator text. + object_id = fields["OBJECT_ID"] + sgp4_fields["OBJECT_ID"] = object_id if object_id[4:5] == "-" else "" satellite = Satrec() # Propagate to epoch to get the True Equator, Mean Equinox (TEME) state vector. @@ -721,6 +720,9 @@ def satOmm2elem(omm_path: str) -> list: including nine-digit OMM IDs, are preserved in ``noradID`` without being constrained by SGP4's internal identifier storage. Values must be in ``[0, 999999999]``; optional leading ``+`` signs and leading zeros are accepted and preserved. + Missing, null or blank international designators yield ``objectID=""``. + Supplied designators, including the literal ``UNKNOWN``, are preserved after + trimming whitespace; backend identifier accommodations do not change this metadata. XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, using either prefixes or a default namespace. Records are read from From b5d6f203844fe74e6d0669edea641b5823efc99a Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:46:17 -0700 Subject: [PATCH 08/15] [#1314] Document and test OMM derivative coefficient conventions --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 48 +++++++++++++++++++ src/utilities/ommHandling.py | 15 +++++- 3 files changed, 62 insertions(+), 3 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index 25ce33d7c0c..ac02b397fdf 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Documents the unscaled mean-motion derivative fields and CelesTrak's TLE coefficient convention. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index ed7f7d1a078..92617df653c 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -1528,6 +1528,54 @@ def test_omm_empty_file_raises(tmp_path): ommHandling.satOmm2elem(str(path)) +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) +@pytest.mark.parametrize("ndot,nddot,tle_ndot,tle_nddot", [ + ("-0.00003657", "1.15630e-5", "-.00003657", " 11563-4"), + ("0.00002182", "-1.2345e-7", " .00002182", "-12345-6"), +]) # Coefficients [rev/day^2], [rev/day^3] in both OMM and TLE representations. +def test_omm_derivative_coefficients_match_tle( + tmp_path, monkeypatch, capsys, encoding, satrec, ndot, nddot, tle_ndot, tle_nddot, +): + """Nonzero signed coefficients retain the TLE convention in metadata and SGP4.""" + from Basilisk.utilities import tleHandling + + line1 = f"1 25544U 98067A 26253.50000000 {tle_ndot} {tle_nddot} 16717-3 0 999" + checksum = sum(int(char) if char.isdigit() else int(char == "-") for char in line1) % 10 + line1 += str(checksum) + line2 = "2 25544 51.6416 247.4627 0004364 130.5360 325.0288 15.50103472478591" + tle_path = tmp_path / "coefficients.2le" + tle_path.write_text(f"{line1}\n{line2}\n", encoding="utf-8") + tle_record = tleHandling.satTle2elem(tle_path)[0] + reference = satrec.twoline2rv(line1, line2) + fields = dict(_OMM_FIELDS, MEAN_MOTION_DOT=ndot, MEAN_MOTION_DDOT=nddot) + path = _WRITERS[encoding](tmp_path / f"coefficients.{encoding}", [fields]) + monkeypatch.setattr(ommHandling, "Satrec", satrec) + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_unscaled_coefficients(satellite, backend_fields): + """Compare backend coefficients against an independently parsed TLE.""" + assert backend_fields["MEAN_MOTION_DOT"] == ndot + assert backend_fields["MEAN_MOTION_DDOT"] == nddot + result = initialize(satellite, backend_fields) + np.testing.assert_allclose( + [satellite.ndot, satellite.nddot], [reference.ndot, reference.nddot], + rtol=1e-14, atol=0.0, + ) + initialized.append(satellite) + return result + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_unscaled_coefficients) + records = ommHandling.satOmm2elem(path) + assert len(records) == len(initialized) == 1 + assert records[0].nDot == float(ndot) + assert records[0].nDotDot == float(nddot) + assert records[0].nDot == pytest.approx(tle_record.nDot, rel=1e-14) + assert records[0].nDotDot == pytest.approx(tle_record.nDotDot, rel=1e-14) + assert capsys.readouterr().out == "" + + def test_omm_matches_tle_for_same_elements(tmp_path): """ The OMM path reproduces the TLE path. diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index fb744c74c31..e673bb27bfd 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -161,6 +161,15 @@ class OmmData: Mirrors :py:class:`Basilisk.utilities.tleHandling.TleData`, with the OMM-only metadata added and the TLE-only line-format fields dropped. + + ``nDot`` and ``nDotDot`` retain the supplied ``MEAN_MOTION_DOT`` and + ``MEAN_MOTION_DDOT`` values without rescaling, in ``rev/day**2`` and ``rev/day**3``. + CelesTrak GP data uses the TLE coefficient convention: one-half of the first + physical derivative and one-sixth of the second physical derivative. + Other producers' conventions must be interpreted using their documentation. + These coefficients are passed unchanged to the SGP4 initializer, which converts + their units; SGP4 itself does not use them for propagation. See + `CelesTrak's TLE field explanation `__. """ # Required (orbit elements and the time at which they are valid) @@ -176,8 +185,8 @@ class OmmData: propagator: str = field(default="0") # EPHEMERIS_TYPE: 0 (CelesTrak default) or 2 (SGP4) elemSetNo: int = field(default=999) # element set number [-] meanMotion: float = field(default=0.0) # mean motion at epoch [rev/day] - nDot: float = field(default=0.0) # first derivative of mean motion [rev/day^2] - nDotDot: float = field(default=0.0) # second derivative of mean motion [rev/day^3] + nDot: float = field(default=0.0) # supplied MEAN_MOTION_DOT [rev/day^2]; see class convention + nDotDot: float = field(default=0.0) # supplied MEAN_MOTION_DDOT [rev/day^3]; see class convention bStar: float = field(default=0.0) # B* drag term [1/Earth radii] centerName: str = field(default="EARTH") # CENTER_NAME refFrame: str = field(default="TEME") # REF_FRAME @@ -723,6 +732,8 @@ def satOmm2elem(omm_path: str) -> list: Missing, null or blank international designators yield ``objectID=""``. Supplied designators, including the literal ``UNKNOWN``, are preserved after trimming whitespace; backend identifier accommodations do not change this metadata. + Mean-motion derivative fields retain the supplied coefficients without rescaling; + see :py:class:`OmmData` for CelesTrak's TLE coefficient convention. XML accepts unqualified elements and the ``urn:ccsds:schema:ndmxml`` namespace, using either prefixes or a default namespace. Records are read from From 26a9c2de3ef2753f4585c3fcd5bdbff3b5bf9677 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 09:48:16 -0700 Subject: [PATCH 09/15] [#1314] Cover representative OMM orbits against TLE states --- src/tests/test_readOMM.py | 50 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 92617df653c..fe1f6355742 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -1166,6 +1166,56 @@ def test_omm_propagates_at_initialized_epoch(tmp_path, capsys, encoding, epoch_s assert capsys.readouterr().out == "" +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) +@pytest.mark.parametrize("mean_motion,eccentricity,inclination,method", [ + pytest.param("15.5", "0.001", "51.6", "n", id="leo"), + pytest.param("2.0", "0.01", "55.0", "d", id="meo"), + pytest.param("1.0027", "0.0001", "0.01", "d", id="geo"), + pytest.param("2.006", "0.72", "63.4", "d", id="molniya"), + pytest.param("1.0027", "0", "0", "d", id="circular-equatorial"), + pytest.param("1.0027", "0", "180", "d", id="circular-retrograde"), + pytest.param("14.0", "0.001", "90.0", "n", id="polar"), +]) # Mean motion [rev/day], eccentricity [-], inclination [deg]; SGP4 branch n/d. +def test_omm_representative_orbit_states( + tmp_path, monkeypatch, capsys, encoding, satrec, mean_motion, eccentricity, inclination, method, +): + """Representative orbits reconstruct independently parsed TLE states at epoch.""" + fields = dict(_OMM_FIELDS, MEAN_MOTION=mean_motion, ECCENTRICITY=eccentricity, INCLINATION=inclination) + path = _WRITERS[encoding](tmp_path / f"orbit.{encoding}", [fields]) + monkeypatch.setattr(ommHandling, "Satrec", satrec) + records = ommHandling.satOmm2elem(path) + assert len(records) == 1 + record = records[0] + epoch = dt.datetime(2026, 9, 10, 12) # UTC + assert record.ommEpoch == epoch + assert record.oe.a > 0.0 # [m] + assert 0.0 <= record.oe.e < 1.0 # [-] + assert all(np.isfinite(getattr(record.oe, name)) for name in ("a", "e", "i", "Omega", "omega", "f")) + + # All selected mean elements are exactly representable at the TLE field precision. + # Initializing through the TLE parser also checks OMM's independent initialization path. + line1 = "1 25544U 98067A 26253.50000000 .00002182 00000-0 16717-3 0 9991" + eccentricity_digits = round(float(eccentricity) * 1e7) + line2 = (f"2 25544 {float(inclination):8.4f} 247.4627 {eccentricity_digits:07d} " + f"130.5360 325.0288 {float(mean_motion):11.8f}47859") + checksum = sum(int(char) if char.isdigit() else int(char == "-") for char in line2) % 10 + line2 += str(checksum) + reference = satrec.twoline2rv(line1, line2) + assert reference.method == method + error, position, velocity = reference.sgp4_tsince(0.0) # [min] since epoch + assert error == 0 + expected_position, expected_velocity = ommHandling._teme2j2000( + np.array(position) * 1e3, np.array(velocity) * 1e3, epoch, # [m], [m/s] + ) + actual_position, actual_velocity = ommHandling.om.elem2rv( + ommHandling.om.MU_EARTH * 1e9, record.oe, # [m^3/s^2] + ) + np.testing.assert_allclose(actual_position, expected_position, rtol=0.0, atol=1e-5) # [m] + np.testing.assert_allclose(actual_velocity, expected_velocity, rtol=0.0, atol=1e-8) # [m/s] + assert capsys.readouterr().out == "" + + def test_omm_bad_record_is_skipped_not_fatal(tmp_path): """A record missing a required element is skipped while good records still return.""" broken = dict(_OMM_FIELDS) From b5ea655b1aef6eee7ec392976e4481cc690cc4d9 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 10:11:08 -0700 Subject: [PATCH 10/15] [#1314] Validate fields throughout OMM XML containers --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 99 +++++++++++++++++++ src/utilities/ommHandling.py | 48 +++++++-- 3 files changed, 139 insertions(+), 10 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index ac02b397fdf..1f0c149bffa 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Documents the unscaled mean-motion derivative fields and CelesTrak's TLE coefficient convention. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement throughout recognized containers, including headers and optional data blocks. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Documents the unscaled mean-motion derivative fields and CelesTrak's TLE coefficient convention. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index fe1f6355742..27c4f47d9a4 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -541,6 +541,105 @@ def test_omm_xml_malformed_structure(tmp_path, capsys, case): assert "XML" in warnings[0] +@pytest.mark.parametrize("destination,namespace_prefix", [ + ("omm", None), ("header", None), ("body", None), + ("spacecraftParameters", None), ("covarianceMatrix", None), ("userDefinedParameters", None), + ("body", "ndm"), ("body", ""), ("spacecraftParameters", "ndm"), ("spacecraftParameters", ""), +]) +def test_omm_xml_outer_field_placement(tmp_path, monkeypatch, capsys, destination, namespace_prefix): + """A misplaced drag field cannot disappear into defaults or hide later records.""" + broken = dict(_OMM_FIELDS, OBJECT_NAME="BROKEN-SAT") + peer = dict(_OMM_FIELDS, OBJECT_NAME="PEER-SAT") + valid = dict(_OMM_FIELDS, OBJECT_NAME="VALID-SAT") + incomplete = {key: value for key, value in broken.items() if key != "MEAN_MOTION"} + path = _writeXml(tmp_path / "outer-field.xml", [broken, peer, valid, incomplete]) + tree = ET.parse(path) + message = tree.find("omm") + body = message.find("body") + other_body = ET.SubElement(ET.SubElement(tree.getroot(), "omm"), "body") + for segment in list(body)[2:]: + body.remove(segment) + other_body.append(segment) + segment = body.find("segment") + parent = segment.find("data/tleParameters") + field = parent.find("BSTAR") + parent.remove(field) + if destination == "omm": + target = message + elif destination == "header": + target = ET.SubElement(message, "header") + elif destination == "body": + target = body + else: + target = ET.SubElement(segment.find("data"), destination) + target.append(field) + tree.write(path, encoding="utf-8") + _set_xml_namespace(path, namespace_prefix) + shared_error = destination in ("omm", "header", "body") + expected_names = ["VALID-SAT"] if shared_error else ["PEER-SAT", "VALID-SAT"] + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, fields): + """Ambiguous messages and malformed segments must not reach SGP4.""" + assert fields["OBJECT_NAME"] in expected_names + initialized.append(fields["OBJECT_NAME"]) + return initialize(satellite, fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == expected_names + assert initialized == expected_names + assert all(record.bStar == float(_OMM_FIELDS["BSTAR"]) for record in records) + warnings = capsys.readouterr().out.splitlines() + indices = [1, 2, 4] if shared_error else [1, 4] + assert len(warnings) == len(indices) + for index, warning in zip(indices, warnings): + assert f"skipped OMM record {index}" in warning + assert ("MEAN_MOTION" if index == 4 else f"Misplaced OMM XML element BSTAR in {destination}") in warning + + +def test_omm_xml_ndm_field_placement(tmp_path, monkeypatch): + """A field outside all OMM messages is a document error, not an optional default.""" + path = _writeXml(tmp_path / "ndm-field.xml", [_OMM_FIELDS, _OMM_FIELDS]) + tree = ET.parse(path) + parameters = tree.find("omm/body/segment/data/tleParameters") + field = parameters.find("BSTAR") + parameters.remove(field) + tree.getroot().append(field) + tree.write(path, encoding="utf-8") + + def unexpected_initialize(*args): + pytest.fail("An invalid NDM document must not reach SGP4") + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", unexpected_initialize) + with pytest.raises(ValueError, match="Misplaced OMM XML element BSTAR in ndm"): + ommHandling.satOmm2elem(path) + + +def test_omm_xml_optional_containers_and_foreign_extensions(tmp_path, capsys): + """Standard optional data and foreign subtrees do not supply consumed OMM fields.""" + path = _writeXml(tmp_path / "optional-containers.xml", [_OMM_FIELDS]) + tree = ET.parse(path) + message = tree.find("omm") + header = ET.SubElement(message, "header") + ET.SubElement(header, "ORIGINATOR").text = "TEST" + data = message.find("body/segment/data") + spacecraft = ET.SubElement(data, "spacecraftParameters") + ET.SubElement(spacecraft, "MASS", units="kg").text = "100" # [kg] + covariance = ET.SubElement(data, "covarianceMatrix") + ET.SubElement(covariance, "COV_REF_FRAME").text = "TEME" + user_defined = ET.SubElement(data, "userDefinedParameters") + ET.SubElement(user_defined, "USER_DEFINED", parameter="BSTAR").text = "unrelated" + for parent, name in ((message, "header"), (data, "spacecraftParameters")): + foreign = ET.SubElement(parent, f"{{urn:foreign}}{name}") + ET.SubElement(foreign, "BSTAR", units="wrong").text = "unrelated" + tree.write(path, encoding="utf-8") + record = ommHandling.satOmm2elem(path)[0] + assert record.bStar == float(_OMM_FIELDS["BSTAR"]) + assert capsys.readouterr().out == "" + + def _write_omm_with_units(path, encoding, field_sets, unit_sets): """Add KVN annotations or XML attributes to selected numeric fields.""" if encoding == "kvn": diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index e673bb27bfd..f7a6ef6c135 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -105,8 +105,22 @@ )), } -#: Recognized structural tags, used to reject blocks in the wrong location. -_OMM_XML_BLOCKS = frozenset(("ndm", "omm", "body", "segment", "data", *_OMM_XML_BLOCK_FIELDS)) +#: Recognized containers to validate recursively; unrelated extension subtrees are ignored. +_OMM_XML_BLOCK_CHILDREN = { + "header": frozenset(), + "body": frozenset(("segment",)), + "data": frozenset(( + "meanElements", "tleParameters", "spacecraftParameters", "covarianceMatrix", + "userDefinedParameters", + )), + **_OMM_XML_BLOCK_FIELDS, + "spacecraftParameters": frozenset(), + "covarianceMatrix": frozenset(), + "userDefinedParameters": frozenset(), +} + +#: Segments are checked separately so a malformed segment does not reject its siblings. +_OMM_XML_BLOCKS = frozenset(("ndm", "omm", "segment", *_OMM_XML_BLOCK_CHILDREN)) #: Fields consumed by this reader; comments and unknown extension fields may repeat. _OMM_FIELDS = frozenset().union(*_OMM_XML_BLOCK_FIELDS.values()) @@ -307,7 +321,10 @@ def _parseOmmJson(text: str) -> list: def _validate_omm_xml_children(block, allowed) -> None: - """Check consumed fields and recognized blocks without interpreting extensions. + """Check recognized containers recursively without interpreting extensions. + + Segments are validated separately to retain per-record error handling. Standard + containers whose fields are not consumed still reject misplaced consumed fields. :param block: XML element whose direct children are being validated :param allowed: field or block names permitted at this location @@ -320,6 +337,8 @@ def _validate_omm_xml_children(block, allowed) -> None: raise ValueError(f"Misplaced OMM XML element {name} in {block.tag}.") if name in _OMM_FIELDS and len(child): raise ValueError(f"OMM XML field {name} must contain text without child elements.") + if name in _OMM_XML_BLOCK_CHILDREN: + _validate_omm_xml_children(child, _OMM_XML_BLOCK_CHILDREN[name]) def _parse_omm_xml_segment(segment) -> dict: @@ -344,12 +363,9 @@ def _parse_omm_xml_segment(segment) -> dict: return duplicates _validate_omm_xml_children(segment, {"metadata", "data"}) - for data in blocks["data"]: - _validate_omm_xml_children(data, {"meanElements", "tleParameters"}) fields = {} - for name, allowed in _OMM_XML_BLOCK_FIELDS.items(): + for name in _OMM_XML_BLOCK_FIELDS: for block in blocks.get(name, []): - _validate_omm_xml_children(block, allowed) for element in block: key = element.tag.upper() if key in _OMM_FIELDS and key in fields: @@ -379,13 +395,23 @@ def _parseOmmXml(text: str) -> list: # Normalize only CCSDS tags so foreign namespaces cannot override OMM fields. for element in root.iter(): element.tag = element.tag.removeprefix("{urn:ccsds:schema:ndmxml}") + if root.tag == "ndm": + _validate_omm_xml_children(root, {"omm"}) messages = [root] if root.tag == "omm" else root.findall("omm") if root.tag == "ndm" else [] records = [] for message in messages: + message_error = None + try: + if len(message.findall("body")) > 1: + raise ValueError("Duplicate OMM XML element: BODY.") + _validate_omm_xml_children(message, {"header", "body"}) + except ValueError as error: + message_error = _OmmInvalidRecord(str(error)) for segment in message.findall("body/segment"): + if message_error is not None: + records.append(message_error) + continue try: - if len(message.findall("body")) != 1: - raise ValueError("Duplicate OMM XML element: BODY.") records.append(_parse_omm_xml_segment(segment)) except ValueError as error: records.append(_OmmInvalidRecord(str(error))) @@ -740,6 +766,10 @@ def satOmm2elem(omm_path: str) -> list: ``omm/body/segment`` in a standalone OMM or a direct OMM child of an NDM container. Segments inside extensions or other message types are ignored. Consumed fields must appear in their owning metadata, mean-elements or TLE-parameters blocks. + This placement check includes headers, bodies and recognized optional containers. + Errors outside a segment invalidate all segments in that OMM message; errors + within a segment invalidate only that record. Other OMM messages remain available. + Misplaced fields or blocks directly under NDM are file-level errors. XML records containing misplaced or duplicate OMM fields or blocks are skipped with a warning before SGP4 is invoked. KVN and JSON records with duplicate consumed fields are also skipped, including From 6bb7c592da2c991f9955372c077d7b7674378038 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 10:12:11 -0700 Subject: [PATCH 11/15] [#1314] Reduce redundant OMM regression test combinations --- src/tests/test_readOMM.py | 91 +++++++++++++++++++++++++-------------- 1 file changed, 58 insertions(+), 33 deletions(-) diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 27c4f47d9a4..9d3a2052a82 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -134,7 +134,7 @@ def _set_xml_namespace(path, prefix, namespace="urn:ccsds:schema:ndmxml"): _WRITERS = {"kvn": _writeKvn, "json": _writeJson, "csv": _writeCsv, "xml": _writeXml} -@pytest.mark.parametrize("encoding", ["kvn", "json", "csv"]) +@pytest.mark.parametrize("encoding", ["kvn", "json"]) @pytest.mark.parametrize("field_name,bad_value", [ ("CENTER_NAME", "MARS"), ("MEAN_MOTION", "nan"), ("OBJECT_NAME", "AMBIGUOUS"), ]) @@ -156,40 +156,45 @@ def test_omm_duplicate_fields( if encoding == "json": record = "{" + ",".join(f"{json.dumps(key)}:{json.dumps(value)}" for key, value in pairs) + "}" text = f"[{record},{json.dumps(_OMM_FIELDS)}]" - elif encoding == "kvn": + else: text = "" for record_pairs in (pairs, list(_OMM_FIELDS.items())): text += "CCSDS_OMM_VERS = 2.0\n" text += "".join(f"{key} = {value}\n" for key, value in record_pairs) - else: - stream = io.StringIO() - writer = csv.writer(stream) - writer.writerow([key for key, value in pairs]) - writer.writerow([value for key, value in pairs]) - text = stream.getvalue() path.write_text(text, encoding="utf-8") initialize = ommHandling.sgp4omm.initialize initialized = [] def initialize_valid_record(satellite, fields): """Only the unambiguous trailing record may reach SGP4.""" - assert encoding != "csv" assert fields["ELEMENT_SET_NO"] == _OMM_FIELDS["ELEMENT_SET_NO"] initialized.append(fields["OBJECT_NAME"]) return initialize(satellite, fields) monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) - if encoding == "csv": - with pytest.raises(ValueError, match=f"Duplicate OMM CSV.*{field_name}"): - ommHandling.satOmm2elem(path) - assert initialized == [] - else: - records = ommHandling.satOmm2elem(path) - assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] - assert initialized == [_OMM_FIELDS["OBJECT_NAME"]] - warning = capsys.readouterr().out - assert "skipped OMM record 1" in warning - assert f"Duplicate OMM {encoding.upper()} field: {field_name}" in warning + records = ommHandling.satOmm2elem(path) + assert [record.satName for record in records] == [_OMM_FIELDS["OBJECT_NAME"]] + assert initialized == [_OMM_FIELDS["OBJECT_NAME"]] + warning = capsys.readouterr().out + assert "skipped OMM record 1" in warning + assert f"Duplicate OMM {encoding.upper()} field: {field_name}" in warning + + +@pytest.mark.parametrize("mixed_case,occurrences", [(False, 2), (True, 2), (False, 3), (True, 3)]) +def test_omm_csv_duplicate_headers(tmp_path, monkeypatch, mixed_case, occurrences): + """CSV rejects ambiguous headers before reading any field values.""" + field_name = "MEAN_MOTION" + duplicate = field_name.lower() if mixed_case else field_name + headers = [*_OMM_FIELDS, *([duplicate] * (occurrences - 1))] + path = tmp_path / "duplicate-headers.csv" + path.write_text(",".join(headers) + "\n", encoding="utf-8") + + def unexpected_initialize(*args): + pytest.fail("An ambiguous CSV header must not reach SGP4") + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", unexpected_initialize) + with pytest.raises(ValueError, match=f"Duplicate OMM CSV.*{field_name}"): + ommHandling.satOmm2elem(path) @pytest.mark.parametrize("bad_first", [False, True]) @@ -471,8 +476,10 @@ def test_omm_xml_unrecognized_root(tmp_path, root_name): ommHandling.satOmm2elem(path) -@pytest.mark.parametrize("namespace_prefix", [None, "ndm", ""]) -@pytest.mark.parametrize("field_name", list(_OMM_FIELDS)) +@pytest.mark.parametrize("namespace_prefix,field_name", [ + *((None, name) for name in _OMM_FIELDS), + *((prefix, name) for prefix in ("ndm", "") for name in ("CENTER_NAME", "MEAN_MOTION", "BSTAR")), +]) @pytest.mark.parametrize("target_offset", [1, 2]) def test_omm_xml_field_block_placement( tmp_path, monkeypatch, capsys, namespace_prefix, field_name, target_offset, @@ -979,12 +986,21 @@ def test_omm_encodings_agree(tmp_path): ), (encoding, name) -@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @pytest.mark.parametrize("catalog_id", [ "00001", "99999", "100000", "100001", "339999", "340000", "999999", "799500001", "999999999", "0", "+0", "+00001", "+25544", "+999999999", "000999999999", ]) +def test_omm_catalog_number_normalization(catalog_id): + """The shared validator preserves accepted decimal spellings at every boundary.""" + fields = dict(_OMM_FIELDS, NORAD_CAT_ID=catalog_id) + normalized = ommHandling._normalizeOmmFields(fields, "kvn") + assert normalized["NORAD_CAT_ID"] == catalog_id + assert fields["NORAD_CAT_ID"] == catalog_id + + +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("catalog_id", ["+00001", "340000", "+999999999"]) @pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id, satrec): """Catalog IDs survive ingestion unchanged without affecting the orbital elements.""" @@ -1006,11 +1022,18 @@ def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id, satrec assert getattr(actual.oe, name) == getattr(reference.oe, name), name -@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @pytest.mark.parametrize("catalog_id", [ "A0001", "100000.5", "-1", "nan", "+", "++25544", "+ 25544", "+-1", "1000000000", "+1000000000", "0001000000000", "١٢٣", "１２３", ]) +def test_omm_invalid_catalog_number_normalization(catalog_id): + """Signs, decimal syntax, Unicode digits and range limits share one validator.""" + with pytest.raises(ValueError, match="NORAD_CAT_ID"): + ommHandling._normalizeOmmFields(dict(_OMM_FIELDS, NORAD_CAT_ID=catalog_id), "kvn") + + +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("catalog_id", ["+", "1000000000", "١٢٣"]) def test_omm_invalid_catalog_number_is_skipped(tmp_path, monkeypatch, capsys, encoding, catalog_id): """Malformed or out-of-range catalog IDs never reach SGP4; later records survive.""" fields = dict(_OMM_FIELDS, NORAD_CAT_ID=catalog_id) @@ -1265,16 +1288,18 @@ def test_omm_propagates_at_initialized_epoch(tmp_path, capsys, encoding, epoch_s assert capsys.readouterr().out == "" -@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) @pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) -@pytest.mark.parametrize("mean_motion,eccentricity,inclination,method", [ - pytest.param("15.5", "0.001", "51.6", "n", id="leo"), - pytest.param("2.0", "0.01", "55.0", "d", id="meo"), - pytest.param("1.0027", "0.0001", "0.01", "d", id="geo"), - pytest.param("2.006", "0.72", "63.4", "d", id="molniya"), - pytest.param("1.0027", "0", "0", "d", id="circular-equatorial"), - pytest.param("1.0027", "0", "180", "d", id="circular-retrograde"), - pytest.param("14.0", "0.001", "90.0", "n", id="polar"), +@pytest.mark.parametrize("encoding,mean_motion,eccentricity,inclination,method", [ + pytest.param("kvn", "15.5", "0.001", "51.6", "n", id="kvn-leo"), + pytest.param("kvn", "2.0", "0.01", "55.0", "d", id="kvn-meo"), + pytest.param("kvn", "1.0027", "0.0001", "0.01", "d", id="kvn-geo"), + pytest.param("kvn", "2.006", "0.72", "63.4", "d", id="kvn-molniya"), + pytest.param("kvn", "1.0027", "0", "0", "d", id="kvn-circular-equatorial"), + pytest.param("kvn", "1.0027", "0", "180", "d", id="kvn-circular-retrograde"), + pytest.param("kvn", "14.0", "0.001", "90.0", "n", id="kvn-polar"), + pytest.param("json", "15.5", "0.001", "51.6", "n", id="json-leo"), + pytest.param("csv", "15.5", "0.001", "51.6", "n", id="csv-leo"), + pytest.param("xml", "15.5", "0.001", "51.6", "n", id="xml-leo"), ]) # Mean motion [rev/day], eccentricity [-], inclination [deg]; SGP4 branch n/d. def test_omm_representative_orbit_states( tmp_path, monkeypatch, capsys, encoding, satrec, mean_motion, eccentricity, inclination, method, From 384572f1fb6b14885096faf311e51e8726530703 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 10:14:36 -0700 Subject: [PATCH 12/15] [#1314] Reuse a writable Matplotlib cache for Python tests --- .../Support/Developer/CodingGuidlines.rst | 8 ++ .../1314-pytest-matplotlib-cache.rst | 1 + src/conftest.py | 24 ++++ src/tests/test_pytest_matplotlib_cache.py | 109 ++++++++++++++++++ 4 files changed, 142 insertions(+) create mode 100644 docs/source/Support/bskReleaseNotesSnippets/1314-pytest-matplotlib-cache.rst create mode 100644 src/tests/test_pytest_matplotlib_cache.py diff --git a/docs/source/Support/Developer/CodingGuidlines.rst b/docs/source/Support/Developer/CodingGuidlines.rst index 9264e024e06..7b274a7897a 100644 --- a/docs/source/Support/Developer/CodingGuidlines.rst +++ b/docs/source/Support/Developer/CodingGuidlines.rst @@ -306,6 +306,14 @@ The default packages include ``pytest`` which is a program that can run a series python scripts that begin with ``test_``. The package ``pytest-xdist`` is also installed by default and allows these tests to be run in a multi-threaded manner using the argument ``-n auto``. +The ``src`` pytest configuration sets ``MPLCONFIGDIR`` to the checkout's +``.pytest_cache/matplotlib`` directory before importing Matplotlib. This ignored, +reusable cache avoids rebuilding the font cache on repeated runs when the home +directory is not writable, and is shared with local parallel test workers and CI +workers. An explicitly supplied ``MPLCONFIGDIR`` is preserved. If the checkout +cache cannot be written, Matplotlib uses its normal cache selection. The first +run after creating or clearing the cache still builds it. + Further, Basilisk uses the Google ``gtest`` suite to run unit tests on C or C++ support libraries. To run all the unit and integrated test open a terminal window and change your working directory to the root ``basilisk`` directory. Activate the python virtual environment if needed. diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-pytest-matplotlib-cache.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-pytest-matplotlib-cache.rst new file mode 100644 index 00000000000..4763f842216 --- /dev/null +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-pytest-matplotlib-cache.rst @@ -0,0 +1 @@ +- Python tests now reuse a writable checkout-local Matplotlib cache to avoid repeated font-cache construction when the home directory is not writable, while preserving an explicit ``MPLCONFIGDIR`` setting. diff --git a/src/conftest.py b/src/conftest.py index 7dbbb6225be..2580af3739e 100644 --- a/src/conftest.py +++ b/src/conftest.py @@ -22,7 +22,31 @@ import shutil import subprocess import sys +import tempfile from datetime import date +from pathlib import Path + + +def _configure_matplotlib_cache(cache_directory): + """Select a reusable, writable test cache unless the caller configured one. + + :param cache_directory: checkout-local directory for Matplotlib configuration and caches + """ + if "MPLCONFIGDIR" in os.environ: + return + try: + cache_directory.mkdir(parents=True, exist_ok=True) + # Check actual writes, including on read-only mounts and Windows. + with tempfile.TemporaryFile(dir=cache_directory): + pass + except OSError: + # A read-only checkout must not prevent Matplotlib's normal fallback. + return + os.environ["MPLCONFIGDIR"] = str(cache_directory) + + +# Configure the cache before importing Matplotlib or pyplot. Workers inherit it. +_configure_matplotlib_cache(Path(__file__).resolve().parents[1] / ".pytest_cache" / "matplotlib") import matplotlib as mpl import matplotlib.pyplot as plt diff --git a/src/tests/test_pytest_matplotlib_cache.py b/src/tests/test_pytest_matplotlib_cache.py new file mode 100644 index 00000000000..9a8e40564d9 --- /dev/null +++ b/src/tests/test_pytest_matplotlib_cache.py @@ -0,0 +1,109 @@ +# ISC License +# +# Copyright (c) 2026, Autonomous Vehicle Systems Lab, University of Colorado at Boulder +# +# Permission to use, copy, modify, and/or distribute this software for any +# purpose with or without fee is hereby granted, provided that the above +# copyright notice and this permission notice appear in all copies. +# +# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES +# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF +# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR +# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES +# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN +# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF +# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + +"""Check persistent Matplotlib cache selection before plotting imports.""" + +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +import conftest +import pytest + + +def test_matplotlib_cache_is_reused(tmp_path, monkeypatch): + """Repeated test sessions select the same directory without deleting its contents.""" + cache = tmp_path / "cache" / "matplotlib" + monkeypatch.delenv("MPLCONFIGDIR", raising=False) + conftest._configure_matplotlib_cache(cache) + assert os.environ["MPLCONFIGDIR"] == str(cache) + cached_file = cache / "cached-data" + cached_file.write_text("preserved", encoding="utf-8") + monkeypatch.delenv("MPLCONFIGDIR") + conftest._configure_matplotlib_cache(cache) + assert os.environ["MPLCONFIGDIR"] == str(cache) + assert cached_file.read_text(encoding="utf-8") == "preserved" + assert list(cache.iterdir()) == [cached_file] + + +@pytest.mark.parametrize("configured", ["custom-cache", ""]) +def test_matplotlib_cache_preserves_environment(tmp_path, monkeypatch, configured): + """An explicit environment setting takes precedence without creating another cache.""" + cache = tmp_path / "unused" + monkeypatch.setenv("MPLCONFIGDIR", configured) + conftest._configure_matplotlib_cache(cache) + assert os.environ["MPLCONFIGDIR"] == configured + assert not cache.exists() + + +@pytest.mark.parametrize("failure", ["directory", "write"]) +def test_matplotlib_cache_unwritable_checkout(tmp_path, monkeypatch, failure): + """Directory creation or write failures leave Matplotlib free to choose its cache.""" + cache = tmp_path / "matplotlib" + monkeypatch.delenv("MPLCONFIGDIR", raising=False) + if failure == "directory": + cache.write_text("a file blocks directory creation", encoding="utf-8") + else: + def denied_write(*args, **kwargs): + raise PermissionError("read-only cache") + + monkeypatch.setattr(conftest.tempfile, "TemporaryFile", denied_write) + conftest._configure_matplotlib_cache(cache) + assert "MPLCONFIGDIR" not in os.environ + + +def test_matplotlib_font_cache_survives_new_processes(tmp_path): + """Fresh Python processes reuse the actual font cache independently of their cwd.""" + configuration = Path(conftest.__file__).resolve() + cache = configuration.parents[1] / ".pytest_cache" / "matplotlib" + try: + cache.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryFile(dir=cache): + pass + except OSError: + pytest.skip("Font-cache integration requires a writable checkout") + + script = """ +import json +import runpy +import sys +from pathlib import Path + +runpy.run_path(sys.argv[1]) +import matplotlib +from matplotlib import font_manager + +cache = Path(matplotlib.get_cachedir()) +print(json.dumps({"directory": str(cache), "files": { + path.name: path.stat().st_mtime_ns for path in cache.glob("fontlist-v*.json") +}})) +""" + environment = os.environ.copy() + environment.pop("MPLCONFIGDIR", None) + environment["MPLBACKEND"] = "Agg" + results = [] + for _ in range(2): + completed = subprocess.run( + [sys.executable, "-c", script, str(configuration)], cwd=tmp_path, + env=environment, check=True, capture_output=True, text=True, + ) + results.append(json.loads(completed.stdout.splitlines()[-1])) + assert results[0]["directory"] == str(cache) + assert results[0]["files"] + assert results[1] == results[0] From 1497db5e640a8b642dbe7b9c5cd7ac421cc6a013 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 10:31:46 -0700 Subject: [PATCH 13/15] [#1314] Support symlinked Matplotlib caches in regression tests --- src/tests/test_pytest_matplotlib_cache.py | 24 ++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/tests/test_pytest_matplotlib_cache.py b/src/tests/test_pytest_matplotlib_cache.py index 9a8e40564d9..201065f131a 100644 --- a/src/tests/test_pytest_matplotlib_cache.py +++ b/src/tests/test_pytest_matplotlib_cache.py @@ -18,6 +18,7 @@ import json import os +import shutil import subprocess import sys import tempfile @@ -68,8 +69,9 @@ def denied_write(*args, **kwargs): assert "MPLCONFIGDIR" not in os.environ -def test_matplotlib_font_cache_survives_new_processes(tmp_path): - """Fresh Python processes reuse the actual font cache independently of their cwd.""" +@pytest.mark.parametrize("symlink_cache", [False, True], ids=["directory", "symlink"]) +def test_matplotlib_font_cache_survives_new_processes(tmp_path, symlink_cache): + """Fresh processes reuse a direct or symlinked font cache independently of their cwd.""" configuration = Path(conftest.__file__).resolve() cache = configuration.parents[1] / ".pytest_cache" / "matplotlib" try: @@ -79,6 +81,22 @@ def test_matplotlib_font_cache_survives_new_processes(tmp_path): except OSError: pytest.skip("Font-cache integration requires a writable checkout") + if symlink_cache: + # Copy the configuration so __file__ identifies an isolated checkout, while + # its cache link shares the persistent cache instead of building a new one. + checkout = tmp_path / "checkout" + source = checkout / "src" + source.mkdir(parents=True) + for name in ("conftest.py", "reportconf.py"): + shutil.copy2(configuration.with_name(name), source / name) + linked_cache = checkout / ".pytest_cache" + try: + linked_cache.symlink_to(cache.parent, target_is_directory=True) + except OSError as error: + pytest.skip(f"Symbolic links are unavailable: {error}") + configuration = source / "conftest.py" + cache = linked_cache / "matplotlib" + script = """ import json import runpy @@ -104,6 +122,6 @@ def test_matplotlib_font_cache_survives_new_processes(tmp_path): env=environment, check=True, capture_output=True, text=True, ) results.append(json.loads(completed.stdout.splitlines()[-1])) - assert results[0]["directory"] == str(cache) + assert Path(results[0]["directory"]).samefile(cache) assert results[0]["files"] assert results[1] == results[0] From 98684c11b65a0c69a4134901bc6ae5b25fbca364 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 12:52:32 -0700 Subject: [PATCH 14/15] [#1314] Accept OMM records without NORAD catalog IDs Preserve missing, null and blank catalog numbers as empty public metadata while retaining validation of supplied IDs and duplicate fields. Add regressions for all four encodings, both SGP4 backends, multiple unidentified records, and OMM 3.0 without catalog metadata. Update the original OMM beta release note. Validation: 2128 related tests passed; focused Sphinx build and pre-commit checks passed. --- .../1314-omm-support.rst | 2 +- src/tests/test_readOMM.py | 105 +++++++++++++++++- src/utilities/ommHandling.py | 37 +++--- 3 files changed, 125 insertions(+), 19 deletions(-) diff --git a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst index 1f0c149bffa..988ce368270 100644 --- a/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst +++ b/docs/source/Support/bskReleaseNotesSnippets/1314-omm-support.rst @@ -1 +1 @@ -- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement throughout recognized containers, including headers and optional data blocks. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Preserves unavailable international designators as empty metadata and retains supplied ``UNKNOWN`` values. Documents the unscaled mean-motion derivative fields and CelesTrak's TLE coefficient convention. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. +- Added :ref:`ommHandling` to read CCSDS Orbit Mean-Elements Message (OMM) general perturbations data in all four CelesTrak encodings (XML, JSON, CSV and KVN) and convert it to osculating orbital elements, supporting catalog numbers through ``999999999`` with optional leading plus signs and preserved leading zeros. XML support includes unqualified elements and CCSDS namespace-qualified elements using prefixes or a default namespace, with per-record rejection of ambiguous duplicate fields and blocks (GitHub issue #1669). Supports standalone OMM and NDM containers, ignores segments inside extensions or other message types, and validates consumed XML field placement throughout recognized containers, including headers and optional data blocks. Rejects ambiguous duplicate consumed fields across all four encodings. CSV support includes leading blank lines and quoted multiline fields, with per-record validation of row widths. Accepts CelesTrak's ``SGP/SGP4`` mean-element theory label as ``SGP4`` across all four encodings (GitHub issue #1673). Accepts records with missing, null or blank catalog numbers without changing their orbital conversion, preserving each unidentified record separately. Preserves unavailable catalog numbers and international designators as empty metadata and retains supplied ``UNKNOWN`` designators. Documents the unscaled mean-motion derivative fields and CelesTrak's TLE coefficient convention. Includes :ref:`scenarioOrbitFromOmm`, a tutorial with bundled synthetic OMM data, spacecraft initialization, and orbit, altitude and speed figures. diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 9d3a2052a82..0105943cae2 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -180,10 +180,13 @@ def initialize_valid_record(satellite, fields): assert f"Duplicate OMM {encoding.upper()} field: {field_name}" in warning -@pytest.mark.parametrize("mixed_case,occurrences", [(False, 2), (True, 2), (False, 3), (True, 3)]) -def test_omm_csv_duplicate_headers(tmp_path, monkeypatch, mixed_case, occurrences): +@pytest.mark.parametrize("field_name,mixed_case,occurrences", [ + ("MEAN_MOTION", False, 2), ("MEAN_MOTION", True, 2), + ("MEAN_MOTION", False, 3), ("MEAN_MOTION", True, 3), + ("NORAD_CAT_ID", True, 2), +]) +def test_omm_csv_duplicate_headers(tmp_path, monkeypatch, field_name, mixed_case, occurrences): """CSV rejects ambiguous headers before reading any field values.""" - field_name = "MEAN_MOTION" duplicate = field_name.lower() if mixed_case else field_name headers = [*_OMM_FIELDS, *([duplicate] * (occurrences - 1))] path = tmp_path / "duplicate-headers.csv" @@ -1022,6 +1025,102 @@ def test_omm_catalog_numbers(tmp_path, monkeypatch, encoding, catalog_id, satrec assert getattr(actual.oe, name) == getattr(reference.oe, name), name +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +@pytest.mark.parametrize("satrec", [ommHandling.Satrec, PythonSatrec], ids=["default", "python"]) +def test_omm_missing_catalog_numbers(tmp_path, monkeypatch, capsys, encoding, satrec): + """Unidentified records retain their metadata, order and distinct orbital states.""" + second = dict(_OMM_FIELDS, OBJECT_NAME="SECOND-SAT", NORAD_CAT_ID="25545", OBJECT_ID="") + second["MEAN_ANOMALY"] = "180.0" # [deg] + identified = [_OMM_FIELDS, second] + unidentified = [{key: value for key, value in fields.items() if key != "NORAD_CAT_ID"} + for fields in identified] + path = _WRITERS[encoding](tmp_path / f"unidentified.{encoding}", unidentified) + reference_path = _WRITERS[encoding](tmp_path / f"identified.{encoding}", identified) + monkeypatch.setattr(ommHandling, "Satrec", satrec) + + records = ommHandling.satOmm2elem(path) + references = ommHandling.satOmm2elem(reference_path) + + assert len(records) == len(references) == 2 + assert [record.noradID for record in records] == ["", ""] + assert [record.satName for record in records] == [fields["OBJECT_NAME"] for fields in identified] + assert [record.objectID for record in records] == [fields["OBJECT_ID"] for fields in identified] + assert records[0].oe.f != records[1].oe.f + for record, reference in zip(records, references): + assert record.ommEpoch == reference.ommEpoch + for name in ("a", "e", "i", "Omega", "omega", "f"): + assert getattr(record.oe, name) == getattr(reference.oe, name), name + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("encoding", ["kvn", "json", "csv", "xml"]) +def test_omm_empty_catalog_numbers(tmp_path, capsys, encoding): + """Blank fields and JSON null preserve missing metadata without inventing an ID.""" + values = ["", " \t "] + ([None] if encoding == "json" else []) + fields = [dict(_OMM_FIELDS, NORAD_CAT_ID=value) for value in values] + path = _WRITERS[encoding](tmp_path / f"empty-catalog.{encoding}", fields) + records = ommHandling.satOmm2elem(path) + assert len(records) == len(values) + assert all(record.noradID == "" for record in records) + assert capsys.readouterr().out == "" + + +def test_omm_data_without_catalog_number(): + """Direct construction also represents an unavailable catalog number as empty.""" + record = ommHandling.OmmData(oe=ommHandling.om.ClassicElements(), ommEpoch=dt.datetime(2026, 9, 10)) + assert record.noradID == "" + + +def test_omm_30_without_catalog_metadata(tmp_path, capsys): + """OMM 3.0 KVN can omit catalog metadata while retaining SGP4 orbital inputs.""" + omitted = {"NORAD_CAT_ID", "EPHEMERIS_TYPE", "CLASSIFICATION_TYPE", "ELEMENT_SET_NO", "REV_AT_EPOCH"} + fields = {key: value for key, value in _OMM_FIELDS.items() if key not in omitted} + path = _writeKvn(tmp_path / "omm-3.0.kvn", [fields, _OMM_FIELDS]) + path.write_text(path.read_text(encoding="utf-8").replace("= 2.0", "= 3.0"), encoding="utf-8") + records = ommHandling.satOmm2elem(path) + assert len(records) == 2 + actual, reference = records + assert actual.noradID == "" + assert actual.classification == "Unclassified" + assert actual.propagator == "0" + assert actual.elemSetNo == 999 + assert actual.revAtEpoch == 0 + for name in ("a", "e", "i", "Omega", "omega", "f"): + assert getattr(actual.oe, name) == getattr(reference.oe, name), name + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("encoding", ["kvn", "json", "xml"]) +@pytest.mark.parametrize("blank_first", [False, True]) +def test_omm_duplicate_empty_catalog_number(tmp_path, monkeypatch, capsys, encoding, blank_first): + """An empty catalog field cannot hide a duplicate in either field order.""" + fields = dict(_OMM_FIELDS, NORAD_CAT_ID="") + path = _WRITERS[encoding](tmp_path / f"duplicate-catalog.{encoding}", [fields, _OMM_FIELDS]) + entries = { + "kvn": ("NORAD_CAT_ID = \n", "NORAD_CAT_ID = 25544\n", ""), + "json": ('"NORAD_CAT_ID": ""', '"NORAD_CAT_ID": "25544"', ", "), + "xml": ("", "25544", ""), + } + blank, supplied, separator = entries[encoding] + duplicate = separator.join([blank, supplied] if blank_first else [supplied, blank]) + path.write_text(path.read_text(encoding="utf-8").replace(blank, duplicate, 1), encoding="utf-8") + initialize = ommHandling.sgp4omm.initialize + initialized = [] + + def initialize_valid_record(satellite, backend_fields): + """Count conversions so an invalid record cannot reach SGP4 unnoticed.""" + initialized.append(backend_fields) + return initialize(satellite, backend_fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert [record.noradID for record in records] == ["25544"] + assert len(initialized) == 1 + warning = capsys.readouterr().out + assert "skipped OMM record 1" in warning + assert "Duplicate OMM" in warning and "NORAD_CAT_ID" in warning + + @pytest.mark.parametrize("catalog_id", [ "A0001", "100000.5", "-1", "nan", "+", "++25544", "+ 25544", "+-1", "1000000000", "+1000000000", "0001000000000", "١٢٣", "１２３", diff --git a/src/utilities/ommHandling.py b/src/utilities/ommHandling.py index f7a6ef6c135..48d61a0c66d 100644 --- a/src/utilities/ommHandling.py +++ b/src/utilities/ommHandling.py @@ -63,7 +63,6 @@ "RA_OF_ASC_NODE", "ARG_OF_PERICENTER", "MEAN_ANOMALY", - "NORAD_CAT_ID", "CLASSIFICATION_TYPE", "EPHEMERIS_TYPE", "ELEMENT_SET_NO", @@ -176,6 +175,8 @@ class OmmData: Mirrors :py:class:`Basilisk.utilities.tleHandling.TleData`, with the OMM-only metadata added and the TLE-only line-format fields dropped. + ``noradID`` is empty when no catalog number is available. + ``nDot`` and ``nDotDot`` retain the supplied ``MEAN_MOTION_DOT`` and ``MEAN_MOTION_DDOT`` values without rescaling, in ``rev/day**2`` and ``rev/day**3``. CelesTrak GP data uses the TLE coefficient convention: one-half of the first @@ -192,7 +193,7 @@ class OmmData: # Optional metadata satName: str = field(default="BSK-Sat-00") # OBJECT_NAME - noradID: str = field(default="00000") # NORAD_CAT_ID; unlike a TLE this may be >= 100000 + noradID: str = field(default="") # NORAD_CAT_ID; empty when unavailable objectID: str = field(default="") # international designator; empty when unavailable classification: str = field(default="Unknown") # "Unclassified", "Classified", "Secret" revAtEpoch: int = field(default=0) # revolution number at epoch [-] @@ -577,11 +578,11 @@ def _normalize_omm_physical_metadata(fields: dict, omm_format: str) -> None: def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: """ - Validate one OMM record and normalize its fields for ``sgp4.omm.initialize``. + Validate one OMM record and normalize its public fields before SGP4 conversion. :param fields: raw OMM field dictionary :param omm_format: detected encoding (``xml``, ``kvn``, ``json``, or ``csv``) - :return: a normalized copy safe to hand to SGP4 + :return: a normalized copy, with unavailable catalog numbers represented by empty strings :raises ValueError: if the record is not a field dictionary or its fields are invalid """ if isinstance(fields, _OmmInvalidRecord): @@ -619,16 +620,19 @@ def _normalizeOmmFields(fields: dict, omm_format: str) -> dict: f"satOmm2elem() found an OMM record missing required field(s): {', '.join(sorted(missing))}" ) - # Validate the external ID before replacing it in the private SGP4 input. - catalog_id = normalized["NORAD_CAT_ID"].strip() - catalog_digits = catalog_id.removeprefix("+") - if not catalog_digits.isascii() or not catalog_digits.isdecimal(): - raise ValueError( - f"satOmm2elem() requires NORAD_CAT_ID to be a non-negative decimal integer: {catalog_id!r}" - ) - # Check the value's range without converting arbitrarily long digit strings to int. - if len(catalog_digits.lstrip("0")) > 9: - raise ValueError(f"OMM NORAD_CAT_ID must be between 0 and 999999999: {catalog_id!r}") + # Catalog numbers identify records but do not affect propagation. Validate supplied + # IDs and preserve missing metadata; the private SGP4 input uses its own identifier. + catalog_id = normalized.get("NORAD_CAT_ID", "").strip() + if catalog_id: + catalog_digits = catalog_id.removeprefix("+") + if not catalog_digits.isascii() or not catalog_digits.isdecimal(): + raise ValueError( + f"satOmm2elem() requires NORAD_CAT_ID to be a non-negative decimal integer: {catalog_id!r}" + ) + # Check the value's range without converting arbitrarily long digit strings to int. + if len(catalog_digits.lstrip("0")) > 9: + raise ValueError(f"OMM NORAD_CAT_ID must be between 0 and 999999999: {catalog_id!r}") + normalized["NORAD_CAT_ID"] = catalog_id _validate_omm_numeric_fields(normalized) _normalize_omm_physical_metadata(normalized, omm_format) @@ -753,8 +757,11 @@ def satOmm2elem(omm_path: str) -> list: The encoding (XML, JSON, CSV or KVN) is detected from the file contents. Catalog numbers, including nine-digit OMM IDs, are preserved in ``noradID`` without being constrained by - SGP4's internal identifier storage. Values must be in ``[0, 999999999]``; + SGP4's internal identifier storage. Supplied values must be in ``[0, 999999999]``; optional leading ``+`` signs and leading zeros are accepted and preserved. + Missing, null or blank catalog numbers yield ``noradID=""`` without affecting + orbital conversion. Each valid input record remains a separate result even when + multiple records have no catalog number. Missing, null or blank international designators yield ``objectID=""``. Supplied designators, including the literal ``UNKNOWN``, are preserved after trimming whitespace; backend identifier accommodations do not change this metadata. From 984774088aea1e2dc06f9462d585c039a0b6ae26 Mon Sep 17 00:00:00 2001 From: Hanspeter Schaub Date: Tue, 10 Nov 2026 13:00:19 -0700 Subject: [PATCH 15/15] [#1314] Cover OMM catalog ID validation boundaries Verify that JSON integer zero remains a catalog ID while false and floating-point zero are rejected before SGP4 initialization. Check that missing catalog IDs do not bypass orbital validation and that later valid records still load. Validation: 2132 related tests passed in 5.00 seconds; pre-commit checks passed. --- src/tests/test_readOMM.py | 57 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/src/tests/test_readOMM.py b/src/tests/test_readOMM.py index 0105943cae2..47102700f5a 100644 --- a/src/tests/test_readOMM.py +++ b/src/tests/test_readOMM.py @@ -1065,6 +1065,63 @@ def test_omm_empty_catalog_numbers(tmp_path, capsys, encoding): assert capsys.readouterr().out == "" +@pytest.mark.parametrize("catalog_id,expected_id", [(0, "0"), (False, None), (0.0, None)], + ids=["integer-zero", "boolean-false", "float-zero"]) +def test_omm_json_zero_catalog_number_types(tmp_path, monkeypatch, capsys, catalog_id, expected_id): + """JSON integer zero is a catalog ID; booleans and floats must fail before SGP4.""" + fields = dict(_OMM_FIELDS, OBJECT_NAME="ZERO-ID-SAT", NORAD_CAT_ID=catalog_id) + path = _writeJson(tmp_path / "zero-catalog.json", [fields, _OMM_FIELDS]) + expected = [(_OMM_FIELDS["OBJECT_NAME"], _OMM_FIELDS["NORAD_CAT_ID"])] + if expected_id is not None: + expected.insert(0, (fields["OBJECT_NAME"], expected_id)) + expected_names = [name for name, _ in expected] + initialize = ommHandling.sgp4omm.initialize + initialized_names = [] + + def initialize_valid_record(satellite, backend_fields): + """Reject any attempt to initialize a record with an invalid catalog ID.""" + assert backend_fields["OBJECT_NAME"] in expected_names + initialized_names.append(backend_fields["OBJECT_NAME"]) + return initialize(satellite, backend_fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert [(record.satName, record.noradID) for record in records] == expected + assert initialized_names == expected_names + warnings = capsys.readouterr().out.splitlines() + if expected_id is None: + assert len(warnings) == 1 + assert "skipped OMM record 1" in warnings[0] + assert "NORAD_CAT_ID" in warnings[0] + else: + assert warnings == [] + + +def test_omm_missing_catalog_number_with_invalid_orbit(tmp_path, monkeypatch, capsys): + """Missing IDs do not bypass orbital validation or prevent later records loading.""" + valid = {key: value for key, value in _OMM_FIELDS.items() if key != "NORAD_CAT_ID"} + broken = dict(valid, OBJECT_NAME="BROKEN-SAT") + broken["ECCENTRICITY"] = "1.0" # [-] outside the supported elliptic-orbit range + path = _writeKvn(tmp_path / "unidentified-invalid-orbit.kvn", [broken, valid]) + initialize = ommHandling.sgp4omm.initialize + initialized_names = [] + + def initialize_valid_record(satellite, backend_fields): + """The invalid unidentified record must be rejected before reaching SGP4.""" + assert backend_fields["OBJECT_NAME"] == valid["OBJECT_NAME"] + initialized_names.append(backend_fields["OBJECT_NAME"]) + return initialize(satellite, backend_fields) + + monkeypatch.setattr(ommHandling.sgp4omm, "initialize", initialize_valid_record) + records = ommHandling.satOmm2elem(path) + assert [(record.satName, record.noradID) for record in records] == [(valid["OBJECT_NAME"], "")] + assert initialized_names == [valid["OBJECT_NAME"]] + warnings = capsys.readouterr().out.splitlines() + assert len(warnings) == 1 + assert "skipped OMM record 1" in warnings[0] + assert "ECCENTRICITY" in warnings[0] + + def test_omm_data_without_catalog_number(): """Direct construction also represents an unavailable catalog number as empty.""" record = ommHandling.OmmData(oe=ommHandling.om.ClassicElements(), ommEpoch=dt.datetime(2026, 9, 10))