Merge remote-tracking branch 'origin/develop' into mueller/pcdu-cmd-updates

This commit is contained in:
Robin Mueller 2022-01-27 16:38:07 +01:00
commit 3a045ec4b4
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
5 changed files with 56 additions and 34 deletions

View File

@ -12,7 +12,7 @@
<option name="ADD_CONTENT_ROOTS" value="true" /> <option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" /> <option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" /> <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtc_client_cli.py" /> <option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtccli.py" />
<option name="PARAMETERS" value="-s ccsds_handler -l -t 8 --hk" /> <option name="PARAMETERS" value="-s ccsds_handler -l -t 8 --hk" />
<option name="SHOW_COMMAND_LINE" value="false" /> <option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" /> <option name="EMULATE_TERMINAL" value="false" />

View File

@ -12,7 +12,7 @@
<option name="ADD_CONTENT_ROOTS" value="true" /> <option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" /> <option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" /> <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtc_client_cli.py" /> <option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtccli.py" />
<option name="PARAMETERS" value="-s 17 -o 0 -t 3 -l" /> <option name="PARAMETERS" value="-s 17 -o 0 -t 3 -l" />
<option name="SHOW_COMMAND_LINE" value="false" /> <option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" /> <option name="EMULATE_TERMINAL" value="true" />

View File

@ -520,10 +520,7 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
"36": ("Star Tracker: Get checksum", {OpCodeDictKeys.TIMEOUT: 2.0}), "36": ("Star Tracker: Get checksum", {OpCodeDictKeys.TIMEOUT: 2.0}),
"37": ("Star Tracker: Set time", {OpCodeDictKeys.TIMEOUT: 2.0}), "37": ("Star Tracker: Set time", {OpCodeDictKeys.TIMEOUT: 2.0}),
"38": ("Star Tracker: Download centroid", {OpCodeDictKeys.TIMEOUT: 2.0}), "38": ("Star Tracker: Download centroid", {OpCodeDictKeys.TIMEOUT: 2.0}),
"39": ( "39": ("Star Tracker: Upload centroid (not implemented?)", {OpCodeDictKeys.TIMEOUT: 2.0}),
"Star Tracker: Upload centroid (not implemented by arcsec?)",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"40": ("Star Tracker: Download matched star", {OpCodeDictKeys.TIMEOUT: 2.0}), "40": ("Star Tracker: Download matched star", {OpCodeDictKeys.TIMEOUT: 2.0}),
"41": ("Star Tracker: Download DB Image", {OpCodeDictKeys.TIMEOUT: 2.0}), "41": ("Star Tracker: Download DB Image", {OpCodeDictKeys.TIMEOUT: 2.0}),
"42": ("Star Tracker: Download Blob Pixel", {OpCodeDictKeys.TIMEOUT: 2.0}), "42": ("Star Tracker: Download Blob Pixel", {OpCodeDictKeys.TIMEOUT: 2.0}),
@ -534,6 +531,10 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
), ),
"45": ("Star Tracker: Upload FPGA image", {OpCodeDictKeys.TIMEOUT: 2.0}), "45": ("Star Tracker: Upload FPGA image", {OpCodeDictKeys.TIMEOUT: 2.0}),
"46": ("Star Tracker: FPGA action", {OpCodeDictKeys.TIMEOUT: 2.0}), "46": ("Star Tracker: FPGA action", {OpCodeDictKeys.TIMEOUT: 2.0}),
"47": ("Star Tracker: Unlock", {OpCodeDictKeys.TIMEOUT: 2.0}),
"48": ("Star Tracker: Request camera parameter", {OpCodeDictKeys.TIMEOUT: 2.0}),
"49": ("Star Tracker: Request limits", {OpCodeDictKeys.TIMEOUT: 2.0}),
"50": ("Star Tracker: Request blob parameters", {OpCodeDictKeys.TIMEOUT: 2.0}),
} }
service_star_tracker_tuple = ("Star tracker", op_code_dict_srv_star_tracker) service_star_tracker_tuple = ("Star tracker", op_code_dict_srv_star_tracker)
@ -542,6 +543,7 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
"1": ("CCSDS Handler: Set high rate", {OpCodeDictKeys.TIMEOUT: 2.0}), "1": ("CCSDS Handler: Set high rate", {OpCodeDictKeys.TIMEOUT: 2.0}),
"2": ("CCSDS Handler: Enable transmitter", {OpCodeDictKeys.TIMEOUT: 2.0}), "2": ("CCSDS Handler: Enable transmitter", {OpCodeDictKeys.TIMEOUT: 2.0}),
"3": ("CCSDS Handler: Disable transmitter", {OpCodeDictKeys.TIMEOUT: 2.0}), "3": ("CCSDS Handler: Disable transmitter", {OpCodeDictKeys.TIMEOUT: 2.0}),
"4": ("CCSDS Handler: Set arbitrary bitrate", {OpCodeDictKeys.TIMEOUT: 2.0}),
} }
service_ccsds_handler_tuple = ("CCSDS Handler", op_code_dict_srv_ccsds_handler) service_ccsds_handler_tuple = ("CCSDS Handler", op_code_dict_srv_ccsds_handler)

View File

@ -5,6 +5,8 @@
@author J. Meier @author J. Meier
@date 20.11.2021 @date 20.11.2021
""" """
import struct
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.packer import TcQueueT from tmtccmd.tc.packer import TcQueueT
from spacepackets.ecss.tc import PusTelecommand from spacepackets.ecss.tc import PusTelecommand
@ -19,44 +21,47 @@ class CommandIds:
EN_TRANSMITTER = bytearray([0x0, 0x0, 0x0, 0x2]) EN_TRANSMITTER = bytearray([0x0, 0x0, 0x0, 0x2])
# Disables the syrlinks transmitter (by using RS485 enables lines) # Disables the syrlinks transmitter (by using RS485 enables lines)
DIS_TRANSMITTER = bytearray([0x0, 0x0, 0x0, 0x3]) DIS_TRANSMITTER = bytearray([0x0, 0x0, 0x0, 0x3])
# Sets an arbitrary bitrate. Normally only set low and set high rate commands should be
# required
ARBITRARY_BITRATE = bytearray([0x0, 0x0, 0x0, 0x4])
def pack_ccsds_handler_test( def pack_ccsds_handler_test(
object_id: bytearray, tc_queue: TcQueueT, op_code: str object_id: bytearray, tc_queue: TcQueueT, op_code: str
) -> TcQueueT: ) -> TcQueueT:
tc_queue.appendleft( tc_queue.appendleft(
( (QueueCommands.PRINT,
QueueCommands.PRINT, "Testing CCSDS handler with object id: 0x" + object_id.hex())
"Testing ccsds handler with object id: 0x" + object_id.hex(),
)
) )
if op_code == "0": if op_code == "0":
tc_queue.appendleft((QueueCommands.PRINT, "CCSDS Handler: Set low rate")) tc_queue.appendleft((QueueCommands.PRINT, "CCSDS Handler: Set low rate"))
command = object_id + CommandIds.SET_LOW_RATE command = object_id + CommandIds.SET_LOW_RATE
command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "1": if op_code == "1":
tc_queue.appendleft((QueueCommands.PRINT, "CCSDS Handler: Set high rate")) tc_queue.appendleft((QueueCommands.PRINT, "CCSDS Handler: Set high rate"))
command = object_id + CommandIds.SET_HIGH_RATE command = object_id + CommandIds.SET_HIGH_RATE
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "2": if op_code == "2":
tc_queue.appendleft( tc_queue.appendleft(
(QueueCommands.PRINT, "CCSDS Handler: Enables the transmitter") (QueueCommands.PRINT, "CCSDS Handler: Enables the transmitter")
) )
command = object_id + CommandIds.EN_TRANSMITTER command = object_id + CommandIds.EN_TRANSMITTER
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "3": if op_code == "3":
tc_queue.appendleft( tc_queue.appendleft(
(QueueCommands.PRINT, "CCSDS Handler: Disables the transmitter") (QueueCommands.PRINT, "CCSDS Handler: Disables the transmitter")
) )
command = object_id + CommandIds.DIS_TRANSMITTER command = object_id + CommandIds.DIS_TRANSMITTER
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "4":
tc_queue.appendleft((QueueCommands.PRINT, "CCSDS Handler: Set arbitrary bitrate"))
bitrate = int(input("Specify bit rate (bps): "))
command = object_id + CommandIds.ARBITRARY_BITRATE + struct.pack('!I', bitrate)
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
return tc_queue return tc_queue

View File

@ -20,6 +20,7 @@ class StarTrackerActionIds:
REQ_VERSION = 2 REQ_VERSION = 2
REQ_INTERFACE = 3 REQ_INTERFACE = 3
REQ_TIME = 4 REQ_TIME = 4
UNLOCK = 6
SWITCH_TO_BOOTLOADER_PROGRAM = 7 SWITCH_TO_BOOTLOADER_PROGRAM = 7
REQ_POWER = 11 REQ_POWER = 11
TAKE_IMAGE = 15 TAKE_IMAGE = 15
@ -58,13 +59,16 @@ class StarTrackerActionIds:
CHANGE_FPGA_DOWNLOAD_FILE = 64 CHANGE_FPGA_DOWNLOAD_FILE = 64
UPLOAD_FPGA_IMAGE = 65 UPLOAD_FPGA_IMAGE = 65
FPGA_ACTION = 66 FPGA_ACTION = 66
REQ_CAMERA_PARAMS = 67
REQ_LIMITS = 68
REQ_BLOB_PARAMS = 69
class ImagePathDefs: class ImagePathDefs:
uploadFile = "/mnt/sd0/startracker/gemma.bin" uploadFile = "/mnt/sd0/startracker/gemma.bin"
downloadFile = "test_image.bin" downloadFile = "test_image.bin"
downloadPath = "/mnt/sd0/startracker" downloadPath = "/mnt/sd0/startracker"
jsonFile = "/mnt/sd0/startracker/test.json" jsonFile = "/mnt/sd0/startracker/full.json"
flashFile = "/mnt/sd0/startracker/flash.bin" flashFile = "/mnt/sd0/startracker/flash.bin"
flashReadPath = "/mnt/sd0/startracker" flashReadPath = "/mnt/sd0/startracker"
uploadCentroidJson = "/mnt/sd0/startracker/upload-centroid.json" uploadCentroidJson = "/mnt/sd0/startracker/upload-centroid.json"
@ -430,26 +434,17 @@ def pack_star_tracker_commands(
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "43": if op_code == "43":
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Download FPGA Image")) tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Download FPGA Image"))
position = 0 position = int(input("Start position: "))
length = 4100 length = int(input("Size to download: "))
command = ( command = object_id + struct.pack('!I', StarTrackerActionIds.DOWNLOAD_FPGA_IMAGE) + \
object_id struct.pack('!I', position) + struct.pack('!I', length) + bytearray(
+ struct.pack("!I", StarTrackerActionIds.DOWNLOAD_FPGA_IMAGE) ImagePathDefs.downloadFpgaImagePath, 'utf-8')
+ struct.pack("!I", position)
+ struct.pack("!I", length)
+ bytearray(ImagePathDefs.downloadFpgaImagePath, "utf-8")
)
command = PusTelecommand(service=8, subservice=128, ssc=66, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=66, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "44": if op_code == "44":
tc_queue.appendleft( tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Change donwload FPGA image file name"))
(QueueCommands.PRINT, "Star tracker: Chnage donwload FPGA image file name") command = object_id + struct.pack('!I', StarTrackerActionIds.CHANGE_FPGA_DOWNLOAD_FILE) + \
) bytearray(ImagePathDefs.downloadFpgaImageName, 'utf-8')
command = (
object_id
+ struct.pack("!I", StarTrackerActionIds.CHANGE_FPGA_DOWNLOAD_FILE)
+ bytearray(ImagePathDefs.downloadFpgaImageName, "utf-8")
)
command = PusTelecommand(service=8, subservice=128, ssc=67, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=67, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "45": if op_code == "45":
@ -471,6 +466,26 @@ def pack_star_tracker_commands(
) )
command = PusTelecommand(service=8, subservice=128, ssc=69, app_data=command) command = PusTelecommand(service=8, subservice=128, ssc=69, app_data=command)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
if op_code == "47":
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Unlock"))
command = object_id + struct.pack('!I', StarTrackerActionIds.UNLOCK)
command = PusTelecommand(service=8, subservice=128, ssc=70, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "48":
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Request camera parameters"))
command = object_id + struct.pack('!I', StarTrackerActionIds.REQ_CAMERA_PARAMS)
command = PusTelecommand(service=8, subservice=128, ssc=71, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "49":
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Request limits"))
command = object_id + struct.pack('!I', StarTrackerActionIds.REQ_LIMITS)
command = PusTelecommand(service=8, subservice=128, ssc=72, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "50":
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Request blob parameters"))
command = object_id + struct.pack('!I', StarTrackerActionIds.REQ_BLOB_PARAMS)
command = PusTelecommand(service=8, subservice=128, ssc=73, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
def pack_write_command(object_id: bytearray) -> bytearray: def pack_write_command(object_id: bytearray) -> bytearray: