kill all the duplication
This commit is contained in:
@ -1,3 +1,5 @@
|
||||
from typing import List
|
||||
|
||||
from config.definitions import CustomServiceList
|
||||
from tmtccmd.config import (
|
||||
QueueCommands,
|
||||
@ -146,12 +148,12 @@ def pack_generic_hk_listening_cmds(
|
||||
)
|
||||
|
||||
elif one_rw is True:
|
||||
activate_one_rw_in_sequence(
|
||||
activate_all_rws_in_sequence(
|
||||
tc_queue=tc_queue,
|
||||
)
|
||||
|
||||
elif two_rws is True:
|
||||
activate_two_rws_in_sequence(
|
||||
activate_all_rws_two_consecutively(
|
||||
tc_queue=tc_queue,
|
||||
)
|
||||
|
||||
@ -652,128 +654,89 @@ def activate_mgts_alternately(
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
||||
|
||||
|
||||
def activate_one_rw_in_sequence(
|
||||
tc_queue: TcQueueT,
|
||||
):
|
||||
def rw_speed_cmd_single(
|
||||
tc_queue: TcQueueT, oid: bytes, init_ssc: int, speed: int, ramp_time: int
|
||||
) -> int:
|
||||
command = pack_set_speed_command(
|
||||
object_id=oid, speed=speed, ramp_time=ramp_time, ssc=init_ssc
|
||||
)
|
||||
init_ssc += 1
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW1_ID, speed=0, ramp_time=ramp_time, ssc=init_ssc
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
return init_ssc + 1
|
||||
|
||||
|
||||
def rw_speed_up_cmd_consec(
|
||||
tc_queue: TcQueueT, obids: List[bytes], init_ssc: int, speed: int, ramp_time: int
|
||||
) -> int:
|
||||
for oid in obids:
|
||||
command = pack_set_speed_command(
|
||||
object_id=oid, speed=speed, ramp_time=ramp_time, ssc=init_ssc
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
init_ssc += 1
|
||||
return init_ssc
|
||||
|
||||
|
||||
def rw_speed_down_cmd_consec(
|
||||
tc_queue: TcQueueT, obids: List[bytes], init_ssc: int, ramp_time: int
|
||||
) -> int:
|
||||
for oid in obids:
|
||||
command = pack_set_speed_command(
|
||||
object_id=oid, speed=0, ramp_time=ramp_time, ssc=init_ssc
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
init_ssc += 1
|
||||
return init_ssc
|
||||
|
||||
|
||||
def activate_all_rws_in_sequence(
|
||||
tc_queue: TcQueueT, init_ssc: int, test_speed: int, test_ramp_time: int
|
||||
) -> int:
|
||||
new_ssc = init_ssc
|
||||
# RW1 speed cmd
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW1_ID,
|
||||
speed=20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_cmd_single(
|
||||
tc_queue, oids.RW1_ID, new_ssc, test_speed, test_ramp_time
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW1_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
||||
|
||||
# RW2 speed cmd
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW2_ID,
|
||||
speed=20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_cmd_single(
|
||||
tc_queue, oids.RW2_ID, new_ssc, test_speed, test_ramp_time
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW2_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
||||
|
||||
# RW3 speed cmd
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW3_ID,
|
||||
speed=20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_cmd_single(
|
||||
tc_queue, oids.RW3_ID, new_ssc, test_speed, test_ramp_time
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW3_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
||||
|
||||
# RW4 speed cmd
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW4_ID,
|
||||
speed=20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_cmd_single(
|
||||
tc_queue, oids.RW4_ID, new_ssc, test_speed, test_ramp_time
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW4_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
||||
return new_ssc
|
||||
|
||||
|
||||
def activate_two_rws_in_sequence(tc_queue: TcQueueT):
|
||||
def activate_all_rws_two_consecutively(tc_queue: TcQueueT, init_ssc: int) -> int:
|
||||
new_ssc = init_ssc
|
||||
# RW1+3 speed cmd
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW1_ID,
|
||||
speed=-20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_up_cmd_consec(
|
||||
tc_queue, [oids.RW1_ID, oids.RW3_ID], new_ssc, -20000, 10000
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW3_ID,
|
||||
speed=-20000,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW1_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW3_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
||||
# RW2+4 speed cmd
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW2_ID,
|
||||
speed=-20000,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_up_cmd_consec(
|
||||
tc_queue, [oids.RW2_ID, oids.RW4_ID], new_ssc, -20000, 10000
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW4_ID,
|
||||
speed=-20000,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW2_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
new_ssc = rw_speed_down_cmd_consec(
|
||||
tc_queue, [oids.RW1_ID, oids.RW3_ID], new_ssc, 10000
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
command = pack_set_speed_command(
|
||||
object_id=oids.RW4_ID,
|
||||
speed=0,
|
||||
ramp_time=10000,
|
||||
)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
||||
new_ssc = rw_speed_down_cmd_consec(
|
||||
tc_queue, [oids.RW2_ID, oids.RW4_ID], new_ssc, 10000
|
||||
)
|
||||
return new_ssc
|
||||
|
Reference in New Issue
Block a user