STR Extensions #244

Merged
muellerr merged 22 commits from str-extensions into main 2023-10-27 12:24:10 +02:00
Showing only changes of commit d9530271c3 - Show all commits

View File

@ -810,6 +810,8 @@ def handle_str_hk_data(set_id: int, hk_data: bytes, pw: PrintWrapper):
pw.dlog(f"Received STR HK set with set ID {set_id}") pw.dlog(f"Received STR HK set with set ID {set_id}")
if set_id == SetId.SOLUTION: if set_id == SetId.SOLUTION:
handle_solution_set(hk_data, pw) handle_solution_set(hk_data, pw)
elif set_id == SetId.HISTOGRAM:
handle_histogram_set(hk_data, pw)
elif set_id == SetId.TEMPERATURE: elif set_id == SetId.TEMPERATURE:
handle_temperature_set(hk_data, pw) handle_temperature_set(hk_data, pw)
elif set_id == SetId.AUTO_BLOB: elif set_id == SetId.AUTO_BLOB:
@ -972,6 +974,7 @@ def handle_blobs_set(hk_data: bytes, pw: PrintWrapper):
pw.dlog("{:<8} {:<8}".format("X", "Y")) pw.dlog("{:<8} {:<8}".format("X", "Y"))
for idx in range(8): for idx in range(8):
pw.dlog("{:<8} {:<8}".format(x_coords[idx], y_coords[idx])) pw.dlog("{:<8} {:<8}".format(x_coords[idx], y_coords[idx]))
assert current_idx == len(hk_data) - 1
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=7) FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=7)
@ -983,7 +986,9 @@ def handle_centroid_set(hk_data: bytes, pw: PrintWrapper):
) )
current_idx = unpack_time_hk(hk_data, 0, pw) current_idx = unpack_time_hk(hk_data, 0, pw)
centroid_count = struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0] centroid_count = struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
current_idx += 4
pw.dlog(f"Centroid count: {centroid_count}") pw.dlog(f"Centroid count: {centroid_count}")
assert current_idx == len(hk_data) - 1
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3) FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3)
@ -993,19 +998,24 @@ def handle_centroids_set(hk_data: bytes, pw: PrintWrapper):
centroids_count = struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0] centroids_count = struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
current_idx += 2 current_idx += 2
pw.dlog(f"Centroids count: {centroids_count}") pw.dlog(f"Centroids count: {centroids_count}")
fmt_coords = "!HHHHHHHHHHHHHHHH" fmt_coords = "!ffffffffffffffff"
inc_len = struct.calcsize(fmt_coords) inc_len = struct.calcsize(fmt_coords)
x_coords = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len]) x_coords = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
y_coords = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len]) y_coords = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
fmt_coords = "!BBBBBBBBBBBBBBBB"
inc_len = struct.calcsize(fmt_coords)
magnitude = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len]) magnitude = struct.unpack(fmt_coords, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
pw.dlog("{:<8} {:<8} {:<8}".format("X", "Y", "Magnitude")) pw.dlog("{:<8} {:<8} {:<8} {:<8}".format("Index", "X", "Y", "Magnitude"))
for idx in range(16): for idx in range(16):
pw.dlog( pw.dlog(
"{:<8} {:<8} {:<8}".format(x_coords[idx], y_coords[idx], magnitude[idx]) "{:<8} {:<8.3f} {:<8.3f} {:<8}".format(
idx, x_coords[idx], y_coords[idx], magnitude[idx]
) )
)
assert current_idx == len(hk_data) - 1
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6) FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6)
@ -1034,13 +1044,13 @@ def handle_matched_centroids_set(hk_data: bytes, pw: PrintWrapper):
y_errors = struct.unpack(fmt_floats, hk_data[current_idx : current_idx + inc_len]) y_errors = struct.unpack(fmt_floats, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
pw.dlog( pw.dlog(
"{:<10} {:<10} {:<10} {:<10} {:<10}".format( "{:<8} {:<10} {:<10} {:<10} {:<10} {:<10}".format(
"Star ID", "X", "Y", "X Error", "Y Error" "Index", "Star ID", "X", "Y", "X Error", "Y Error"
) )
) )
for idx in range(16): for idx in range(16):
pw.dlog( pw.dlog(
"{:<10} {:<10} {:<10.3f} {:<10.3f} {:<10.3f} {:<10.3f}".format( "{:<8} {:<10} {:<10.3f} {:<10.3f} {:<10.3f} {:<10.3f}".format(
idx, idx,
star_id[idx], star_id[idx],
x_coords[idx], x_coords[idx],
@ -1049,6 +1059,7 @@ def handle_matched_centroids_set(hk_data: bytes, pw: PrintWrapper):
y_errors[idx], y_errors[idx],
) )
) )
assert current_idx == len(hk_data) - 1
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=8) FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=8)
@ -1065,9 +1076,50 @@ def handle_auto_blob_set(hk_data: bytes, pw: PrintWrapper):
fmt_threshold, hk_data[current_idx : current_idx + inc_len] fmt_threshold, hk_data[current_idx : current_idx + inc_len]
)[0] )[0]
pw.dlog(f"Threshold {threshold}") pw.dlog(f"Threshold {threshold}")
assert current_idx == len(hk_data) - 1
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3) FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3)
def handle_histogram_set(hk_data: bytes, pw: PrintWrapper):
pw.dlog("Received Histogram Set")
current_idx = unpack_time_hk(hk_data, 0, pw)
fmt_str = "!IIIIIIIII"
bins_list = []
inc_len = struct.calcsize(fmt_str)
a_bins = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
bins_list.append(a_bins)
current_idx += inc_len
b_bins = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
bins_list.append(b_bins)
current_idx += inc_len
c_bins = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
bins_list.append(c_bins)
current_idx += inc_len
d_bins = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
bins_list.append(d_bins)
pw.dlog(
"{:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8}".format(
"Section", "Index", "0", "1", "2", "3", "4", "5", "6", "7", "8"
)
)
for idx in range(4):
name = "?"
if idx == 0:
name = "A"
if idx == 1:
name = "B"
if idx == 2:
name = "C"
if idx == 3:
name = "D"
pw.dlog(
"{:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8} {:<8}".format(
name, idx, *bins_list[idx]
)
)
pass
def handle_star_tracker_action_replies( def handle_star_tracker_action_replies(
action_id: int, pw: PrintWrapper, custom_data: bytes action_id: int, pw: PrintWrapper, custom_data: bytes
): ):