Use different alg for two_byte_sum that fixes off-by-one error (#315)

* Use different alg for two_byte_sum that fixes off-by-one error

* ruff format 0.9

* Make fix that works for masked arrays

* Add a unit test

* Update test

Author: taldcroft

mica/archive/aca_dark/dark_cal.py

@@ -141,7 +141,7 @@ def _get_dark_cal_id_scalar(date, select="before", dark_cal_ids=None):
     if ii < 0:
         earliest = CxoTime(dark_cal_secs[0]).date[:8]
         raise MissingDataError(
-            f"No dark cal found before {earliest}" f"(requested dark cal on {date})"
+            f"No dark cal found before {earliest}(requested dark cal on {date})"
         )
 
     try:

mica/archive/aca_hdr3.py

@@ -23,12 +23,21 @@
 
 
 def two_byte_sum(byte_msids, scale=1):
-    def func(slot_data):
-        return (
-            (slot_data[byte_msids[0]].astype("int") >> 7) * (-1 * 65535)
-            + (slot_data[byte_msids[0]].astype("int") << 8)
-            + (slot_data[byte_msids[1]].astype("int"))
-        ) * scale
+    def func(slot_data) -> np.ma.MaskedArray:
+        # For each pair bytes0[i], bytes1[i], return the 16-bit signed integer
+        # corresponding to those two bytes. The input bytes are unsigned.
+        bytes0 = slot_data[byte_msids[0]].astype(np.uint8)
+        bytes1 = slot_data[byte_msids[1]].astype(np.uint8)
+
+        # Make a 2xN array, then transpose to Nx2, then flatten to 2N, then copy to
+        # get values continous in memory.
+        bytes8_2xN = np.ma.vstack([bytes0, bytes1], dtype=np.uint8)
+        bytes8 = bytes8_2xN.transpose().flatten().copy()
+
+        # Now view the 2N bytes as N 16-bit signed integers.
+        ints16 = np.ma.array(bytes8.data.view(">i2"), mask=bytes8.mask[::2])
+
+        return ints16 * scale
 
     return func
 

mica/archive/cda/services.py

@@ -430,8 +430,7 @@ def _get_cda_service_text(service, timeout=60, **params):
 
     if not resp.ok:
         raise RuntimeError(
-            f"got error {resp.status_code} for {resp.url}\n"
-            f"{html_to_text(resp.text)}"
+            f"got error {resp.status_code} for {resp.url}\n{html_to_text(resp.text)}"
         )
 
     return resp.text
@@ -630,7 +629,7 @@ def get_ocat_local(
         # accurate enough for this application.
         where = (
             f"arccos(sin({ra * d2r})*sin(ra*{d2r}) + "
-            f"cos({ra * d2r})*cos(ra*{d2r})*cos({dec*d2r}-dec*{d2r}))"
+            f"cos({ra * d2r})*cos(ra*{d2r})*cos({dec * d2r}-dec*{d2r}))"
             f"< {radius / 60 * d2r}"
         )
         where_parts.append(where)

mica/archive/tests/test_aca_hdr3.py

@@ -7,6 +7,7 @@
 
 import numpy as np
 import pytest
+from astropy.table import Table
 
 from mica.archive import aca_hdr3
 
@@ -63,3 +64,24 @@ def test_MSIDset():
             10679,
         ]
     )
+
+
+def test_two_byte_sum():
+    bytes0 = np.ma.array([0x00, 0xF0, 0x0F, 0xFF, 0xFF], dtype=np.uint8)
+    bytes1 = np.ma.array([0x00, 0x0F, 0xF0, 0xFF, 0xFF], dtype=np.uint8)
+    bytes0[-1] = np.ma.masked
+    bytes1[-1] = np.ma.masked
+
+    # Original code prior to PR #315
+    out1 = (
+        (bytes0.astype("int") >> 7) * (-1 * 65535)
+        + (bytes0.astype("int") << 8)
+        + (bytes1.astype("int"))
+    )
+    assert np.all(out1 == np.ma.array([0, -4080, 4080, 0, 0], mask=[0, 0, 0, 0, 1]))
+
+    # New code in PR #315
+    slot_data = Table([bytes0, bytes1], names=["byte0", "byte1"])
+    ints16 = aca_hdr3.two_byte_sum(["byte0", "byte1"])(slot_data)
+
+    assert np.all(ints16 == np.ma.array([0, -4081, 4080, -1, 0], mask=[0, 0, 0, 0, 1]))

mica/centroid_dashboard.py

@@ -1182,7 +1182,7 @@ def make_html(row_obsid, rows_slot, obs_dir):
 
     for row in t_slot:
         string += f"""<tr>
-<td align='right'>{row['slot']}</td>
+<td align='right'>{row["slot"]}</td>
 """
         if row["id"] < 100:
             id_ = ""
@@ -1193,13 +1193,13 @@ def make_html(row_obsid, rows_slot, obs_dir):
             )
 
         string += f"""<td align='right'>{id_}</td>
-<td align='right'>{row['type']}</td>
-<td align='right'>{row['mag']}</td>
-<td align='right'>{row['yang']}</td>
-<td align='right'>{row['zang']}</td>
-<td align='right'>{row['median_mag']:.3f}</td>
-<td align='right'>{row['median_dy']:.2f}</td>
-<td align='right'>{row['median_dz']:.2f}</td>
+<td align='right'>{row["type"]}</td>
+<td align='right'>{row["mag"]}</td>
+<td align='right'>{row["yang"]}</td>
+<td align='right'>{row["zang"]}</td>
+<td align='right'>{row["median_mag"]:.3f}</td>
+<td align='right'>{row["median_dy"]:.2f}</td>
+<td align='right'>{row["median_dz"]:.2f}</td>
 </tr>
 """
     string += f"""</table>

mica/report/report.py

@@ -920,9 +920,9 @@ def save_state_in_db(obsid, notes):
     del notes["last_sched"]
 
     idcheck = db.fetchone(
-        "select * from report_proc "
-        "where obsid = '{}' "
-        "and report_version = '{}'".format(obsid, REPORT_VERSION)
+        "select * from report_proc where obsid = '{}' and report_version = '{}'".format(
+            obsid, REPORT_VERSION
+        )
     )
 
     if idcheck is None:

mica/stats/update_acq_stats.py

@@ -126,8 +126,7 @@ def get_options():
     parser.add_argument(
         "--email",
         action="append",
-        help="email warning recipient, specify multiple times "
-        "for multiple recipients",
+        help="email warning recipient, specify multiple times for multiple recipients",
     )
     opt = parser.parse_args()
     return opt

mica/vv/process.py

@@ -158,7 +158,7 @@ def update(obsids=None):
                 logger.warn(f"Skipping obs:ver {obsid}:{obs['revision']}. Missing data")
                 continue
             update_str = f"""UPDATE aspect_1_proc set vv_complete = {VV_VERSION}
-                              where obsid = {obsid} and revision = {obs['revision']}"""
+                              where obsid = {obsid} and revision = {obs["revision"]}"""
 
             logger.info(update_str)
             with ska_dbi.DBI(dbi="sqlite", server=FILES["asp1_proc_table"]) as db:

mica/web/views.py

@@ -83,7 +83,7 @@ def get_context_data(self, **kwargs):
                 context["gui_table"] = gui_table
                 reports_url = (
                     "https://cxc.cfa.harvard.edu/mta/ASPECT/agasc/supplement_reports/stars/"
-                    f"{int(agasc_id//1e7):03d}/{agasc_id}/index.html"
+                    f"{int(agasc_id // 1e7):03d}/{agasc_id}/index.html"
                 )
                 context["reports_url"] = reports_url
         return context

scripts/update_agasc_supplement.py

@@ -38,8 +38,7 @@ def get_options(args=None):
         "--bad-star-source",
         type=int,
         help=(
-            "Source identifier indicating provenance (default=max "
-            "existing source + 1)"
+            "Source identifier indicating provenance (default=max existing source + 1)"
         ),
     )
     parser.add_argument(