Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- Fixed :ref:`ommHandling` XML reading to accept CCSDS namespace-qualified OMM records while preserving existing unqualified XML support and validation.
95 changes: 95 additions & 0 deletions src/tests/test_readOMM.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,101 @@ def block(name, keys):
return path


def _qualifyOmmXml(path, namespace="urn:ccsds:schema:ndmxml"):
"""Rewrite the existing XML fixture with namespace-qualified CCSDS tags."""
tree = ET.parse(path)
for element in tree.iter():
element.tag = f"{{{namespace}}}{element.tag}"
tree.write(path, encoding="utf-8", xml_declaration=True)
return path


def test_omm_xml_qualified_namespace_extracts_all_fields(tmp_path):
"""A CCSDS-qualified record must not vanish before field validation."""
path = _qualifyOmmXml(_writeXml(tmp_path / "qualified.xml", [_OMM_FIELDS]))

records = ommHandling._parseOmmXml(path.read_text(encoding="utf-8"))

assert len(records) == 1
assert records[0] == _OMM_FIELDS


def test_omm_xml_qualified_namespace_matches_unqualified_orbit(tmp_path):
"""Equivalent supported OMM XML encodings must initialize one orbit."""
path = _writeXml(tmp_path / "orbit.xml", [_OMM_FIELDS])
original = ommHandling.satOmm2elem(str(path))
_qualifyOmmXml(path)

qualified = ommHandling.satOmm2elem(str(path))

assert len(qualified) == len(original) == 1
assert qualified[0].noradID == "25544"
assert qualified[0].ommEpoch == original[0].ommEpoch
assert abs(qualified[0].oe.a - original[0].oe.a) <= A_TOL_DIST
assert abs(qualified[0].oe.e - original[0].oe.e) <= A_TOL
for angle in ("i", "Omega", "omega", "f"):
assert abs(getattr(qualified[0].oe, angle) - getattr(original[0].oe, angle)) <= A_TOL


@pytest.mark.parametrize("units,expected_count", [("rev/day", 1), ("km", 0)])
def test_omm_xml_qualified_numeric_units_are_validated(tmp_path, units, expected_count):
"""Qualified XML keeps the same numeric-unit acceptance boundary."""
path = _writeXml(tmp_path / "units.xml", [_OMM_FIELDS])
tree = ET.parse(path)
tree.find(".//MEAN_MOTION").set("units", units)
tree.write(path, encoding="utf-8")
_qualifyOmmXml(path)

records = ommHandling.satOmm2elem(str(path))

assert len(records) == expected_count


def test_omm_xml_other_namespace_not_accepted(tmp_path):
"""An unrelated XML namespace is not interpreted as CCSDS OMM."""
path = _qualifyOmmXml(_writeXml(tmp_path / "other.xml", [_OMM_FIELDS]),
namespace="urn:example:not-ccsds")

assert ommHandling._parseOmmXml(path.read_text(encoding="utf-8")) == []


def test_omm_xml_bare_segment_root_not_accepted():
"""A bare segment is not a complete OMM document."""
xml = "<segment><metadata><OBJECT_NAME>TEST</OBJECT_NAME></metadata></segment>"

assert ommHandling._parseOmmXml(xml) == []


def test_omm_xml_qualified_and_unqualified_segments_keep_order():
"""Record order must be preserved if both accepted tag forms appear."""
ns = "urn:ccsds:schema:ndmxml"
root = ET.Element("ndm")
for tag, name in (("segment", "FIRST"), (f"{{{ns}}}segment", "SECOND")):
segment = ET.SubElement(root, tag)
metadata = ET.SubElement(segment, tag.replace("segment", "metadata"))
ET.SubElement(metadata, tag.replace("segment", "OBJECT_NAME")).text = name

records = ommHandling._parseOmmXml(ET.tostring(root, encoding="unicode"))

assert [record["OBJECT_NAME"] for record in records] == ["FIRST", "SECOND"]


def test_omm_xml_qualified_incomplete_record_keeps_number(tmp_path, capsys):
"""Missing fields in a qualified record still report the correct index."""
path = _writeXml(tmp_path / "incomplete.xml", [_OMM_FIELDS] * 3)
tree = ET.parse(path)
second = tree.findall(".//segment")[1]
mean_elements = second.find(".//meanElements")
mean_elements.remove(mean_elements.find("EPOCH"))
tree.write(path, encoding="utf-8")
_qualifyOmmXml(path)

records = ommHandling.satOmm2elem(str(path))

assert len(records) == 2
assert "skipped OMM record 2" in capsys.readouterr().out


_WRITERS = {"kvn": _writeKvn, "json": _writeJson, "csv": _writeCsv, "xml": _writeXml}


Expand Down
20 changes: 15 additions & 5 deletions src/utilities/ommHandling.py
Original file line number Diff line number Diff line change
Expand Up @@ -253,31 +253,41 @@ def _parseOmmJson(text: str) -> list:
return payload


_CCSDS_XML_NAMESPACE = "{urn:ccsds:schema:ndmxml}" # CCSDS 505.0-B-2 qualified XML tag prefix


def _parseOmmXml(text: str) -> list:
"""
Parse the XML (CCSDS NDM) encoding.

Retain the available fields of every segment, including those missing metadata
or data, so per-record validation reports their original positions in the file.
Accept both unqualified tags and the CCSDS NDM namespace-qualified tag form.

:param text: full text of the XML file
:return: list of raw field dictionaries retaining numeric unit declarations
"""
root = ET.fromstring(text)
records = []
for segment in root.findall(".//segment"):
metadata = segment.find("metadata")
data = segment.find("data")
for segment in root.iter():
if segment is root or segment.tag not in ("segment", f"{_CCSDS_XML_NAMESPACE}segment"):
continue
namespace = _CCSDS_XML_NAMESPACE if segment.tag.startswith("{") else ""
metadata = segment.find(f"{namespace}metadata")
data = segment.find(f"{namespace}data")

fields = {}
blocks = [metadata]
if data is not None:
blocks.extend([data.find("meanElements"), data.find("tleParameters")])
blocks.extend([data.find(f"{namespace}meanElements"),
data.find(f"{namespace}tleParameters")])
for block in blocks:
if block is None:
continue
for element in block:
key = element.tag.upper()
if namespace and not element.tag.startswith(namespace):
continue
key = element.tag[len(namespace):].upper()
value = (element.text or "").strip()
if key in _OMM_FIELD_UNITS and "units" in element.attrib:
value = _OmmFieldWithUnits(value, element.attrib["units"])
Expand Down