Merge pull request #321 from NASA-AMMOS/issue-320

Issue #320 - Add limits_find_dn utility

Author: MJJoyce

ait/core/bin/ait_limits_find_dn.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python
+# Advanced Multi-Mission Operations System (AMMOS) Instrument Toolkit (AIT)
+# Bespoke Link to Instruments and Small Satellites (BLISS)
+#
+# Copyright 2021, by the California Institute of Technology. ALL RIGHTS
+# RESERVED. United States Government Sponsorship acknowledged. Any
+# commercial use must be negotiated with the Office of Technology Transfer
+# at the California Institute of Technology.
+#
+# This software may be subject to U.S. export control laws. By accepting
+# this software, the user agrees to comply with all applicable U.S. export
+# laws and regulations. User has the responsibility to obtain export licenses,
+# or other export authority as may be required before exporting such
+# information to foreign countries or providing access to foreign persons.
+
+'''
+ait-limits-find-dn
+
+Extract the DN-equivalent value for EU limit trip thresholds
+
+usage:
+    > ait-limits-find-dn
+
+This utility requires well-formed telemetry and limits dictionaries. DN thresholds
+are brute forced for each limit. Partial limit definitions (i.e., without an
+upper and lower error / warn value) are supported. Any missing limit values or
+points where a threshold cannot be found for the field's DN-to-EU equation will
+be displayed as None in the final table.
+
+Note, "value" limits are not supported by this script and will be skipped. An
+example value limit:
+
+    - !Limit
+      source: Ethernet_HS_Packet.product_type
+      desc: Ethernet Product Type field
+      value:
+        error: MEM_DUMP
+        warn:
+          - TABLE_FOO
+          - TABLE_BAR
+'''
+
+from ait.core import limits
+from ait.core import log
+from ait.core import tlm
+
+
+def main():
+    DNLimits = {}
+    EULimits = {}
+    EUValues = {}
+
+    ld = limits.getDefaultDict()
+    td = tlm.getDefaultDict()
+
+    all_vals = [
+        [
+            "Telem Point",
+            "EU lower.error",
+            "EU lower.warn",
+            "EU upper.warn",
+            "EU upper.error",
+            "DN lower.error",
+            "DN lower.warn",
+            "DN upper.warn",
+            "DN upper.error",
+            "DN to EU LE",
+            "DN to EU LW",
+            "DN to EU UW",
+            "DN to EU UE",
+        ]
+    ]
+
+    for source in sorted(ld.keys(), key=lambda x: x.split(".")[1]):
+        log.info(f"Processing {source}")
+        limit = ld[source]
+        pkt_name, name = source.split(".")
+
+        # Don't support limits specifying individual values. This is usually
+        # used to specify enumerations that aren't valid and we don't properly
+        # handle those cases.
+        if ld[source].value is not None:
+            log.warn(f'Skipping unsupported "value" limit {source}')
+            continue
+
+        DNLimits.setdefault(name, [None, None, None, None])
+        EULimits.setdefault(name, [None, None, None, None])
+        EUValues.setdefault(name, [None, None, None, None])
+
+        if ld[source].lower is not None:
+            try:
+                EULimits[name][0] = ld[source].lower.error
+            except AttributeError:
+                pass
+
+            try:
+                EULimits[name][1] = ld[source].lower.warn
+            except AttributeError:
+                pass
+
+        if ld[source].upper is not None:
+            try:
+                EULimits[name][2] = ld[source].upper.warn
+            except AttributeError:
+                pass
+
+            try:
+                EULimits[name][3] = ld[source].upper.error
+            except AttributeError:
+                pass
+
+        values = []
+
+        defn = td[pkt_name]
+        data = bytearray(defn.nbytes)
+        packet = tlm.Packet(defn, data)
+        for dn in range(65536):
+            setattr(packet, name, dn)
+            eu = getattr(packet, name)
+
+            if eu is not None:
+                values.append((dn, eu))
+
+        values.sort(key=lambda pair: pair[1])
+
+        for dn, eu in values:
+            for n in range(4):
+                if (
+                    EULimits[name][n] is not None
+                    and DNLimits[name][n] is None
+                    and eu > EULimits[name][n]
+                ):
+                    value = dn - 1 if dn > 0 else 0
+                    DNLimits[name][n] = value
+
+                    setattr(packet, name, value)
+                    EUValues[name][n] = getattr(packet, name)
+
+            if all(DNLimits[name][n] is not None for n in range(4)):
+                break
+
+        values = [source]
+        values.extend(map(str, EULimits[name]))
+        values.extend(map(str, DNLimits[name]))
+        values.extend(map(str, EUValues[name]))
+        all_vals.append(values)
+
+    s = [[str(e) for e in row] for row in all_vals]
+    lens = [max(map(len, col)) for col in zip(*s)]
+    fmt = "\t".join("{{:{}}}".format(x) for x in lens)
+    table = [fmt.format(*row) for row in s]
+    print("\n".join(table))
+
+if __name__ == '__main__':
+    main()

doc/source/ait.core.bin.ait_limits_find_dn.rst

@@ -0,0 +1,7 @@
+ait.core.bin.ait\_limits\_find\_dn module
+=========================================
+
+.. automodule:: ait.core.bin.ait_limits_find_dn
+   :members:
+   :undoc-members:
+   :show-inheritance:

doc/source/ait.core.bin.rst

@@ -14,6 +14,7 @@ Submodules
    ait.core.bin.ait_cmd_send
    ait.core.bin.ait_create_dirs
    ait.core.bin.ait_dict_writer
+   ait.core.bin.ait_limits_find_dn
    ait.core.bin.ait_mps_seq_convert
    ait.core.bin.ait_pcap
    ait.core.bin.ait_pcap_segment

doc/source/command_line.rst

@@ -120,6 +120,12 @@ ait-create-dirs
    :start-after: '''
    :end-before: '''
 
+ait-limits-find-dn
+^^^^^^^^^^^^^^^^^^
+.. literalinclude:: ../../ait/core/bin/ait_limits_find_dn.py
+   :start-after: '''
+   :end-before: '''
+
 ait-yaml-validate
 ^^^^^^^^^^^^^^^^^^^
 .. literalinclude:: ../../ait/core/bin/ait_yaml_validate.py