01234567890123456789012345678901234567890123456789012345678901234567890123456789
605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645 1163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238 | <----SKIPPED LINES----> def ServoMain(to_arduino_q, to_parent_q, shutdown): """Main servo controller for projecting the plane position on a hemisphere. Takes the latest flight from the to_arduino_q and converts that to the current azimuth and altitude of the plane on a hemisphere. """ sys.stderr = open(messageboard.STDERR_FILE, 'a') Log('Process started with process id %d' % os.getpid()) # Ensures that the child can exit if the parent exits unexpectedly # docs.python.org/2/library/multiprocessing.html#multiprocessing.Queue.cancel_join_thread to_arduino_q.cancel_join_thread() to_parent_q.cancel_join_thread() # write_format: azimuth, altitude, R, G, & B intensity # read heartbeat: millis link = Serial( *SERVO_CONNECTION, read_timeout=5, error_pin=messageboard.GPIO_ERROR_ARDUINO_SERVO_CONNECTION, to_parent_q=to_parent_q, read_format='l', write_format='ffhhh', name='Servo') link.Open() last_flight = {} last_angles = (0, 0) flight, json_desc_dict, configuration, additional_attr = InitialMessageValues( to_arduino_q) next_read = 0 next_write = 0 now = GetNow(json_desc_dict, additional_attr) flight_present = False while not shutdown.value: if not to_arduino_q.empty(): flight, json_desc_dict, configuration, additional_attr = to_arduino_q.get( block=False) if 'test_servos' in configuration: messageboard.RemoveSetting(configuration, 'test_servos') <----SKIPPED LINES----> ('setting_off_time', 'H'), ('setting_delay', 'H'), ('last_plane', '?'), ('display_mode', 'H'), ('histogram_enabled', '?'), ('current_hist_type', 'H'), ('current_hist_history', 'H'), ('low_battery', '?')) write_config = ( ('setting_screen_enabled', '?'), # 1 bytes ('setting_max_distance', 'H'), # 2 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_on_time', 'H'), # 2 bytes ('setting_off_time', 'H'), # 2 bytes ('setting_delay', 'H'), # 2 bytes ('line1', '9s'), # 9 bytes; 8 character plus terminator ('line2', '9s'), # 9 bytes; 8 character plus terminator ('line1_dec_mask', 'H'), # 2 bytes ('line2_dec_mask', 'H'), # 2 bytes # testing sending a long message - extra 80 bytes - to see if failure time changes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_max_altitude', 'L') # 4 bytes ) #pylint: enable = bad-whitespace read_keys, unused_read_format_tuple, read_format_string = SplitFormat(read_config) write_keys, write_format_tuple, write_format_string = SplitFormat(write_config) values_d = {} low_batt = False to_parent_q.put(('pin', (messageboard.GPIO_ERROR_BATTERY_CHARGE, low_batt))) link = Serial( *REMOTE_CONNECTION, read_timeout=5, error_pin=messageboard.GPIO_ERROR_ARDUINO_REMOTE_CONNECTION, to_parent_q=to_parent_q, read_format=read_format_string, write_format=write_format_string, name='Remote') link.Open() display_mode = 2 #TODO: remove after switch integration flight, json_desc_dict, configuration, additional_attr = InitialMessageValues( to_arduino_q) next_read = 0 next_write = 0 while not shutdown.value: if not to_arduino_q.empty(): to_arduino_message = to_arduino_q.get(block=False) flight, json_desc_dict, configuration, additional_attr = to_arduino_message if 'test_remote' in configuration: messageboard.RemoveSetting(configuration, 'test_remote') def TestDisplayMode(m): <----SKIPPED LINES----> |
01234567890123456789012345678901234567890123456789012345678901234567890123456789
605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645 11631164116511661167116811691170117111721173117411751176117711781179118011811182 1183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213 | <----SKIPPED LINES----> def ServoMain(to_arduino_q, to_parent_q, shutdown): """Main servo controller for projecting the plane position on a hemisphere. Takes the latest flight from the to_arduino_q and converts that to the current azimuth and altitude of the plane on a hemisphere. """ sys.stderr = open(messageboard.STDERR_FILE, 'a') Log('Process started with process id %d' % os.getpid()) # Ensures that the child can exit if the parent exits unexpectedly # docs.python.org/2/library/multiprocessing.html#multiprocessing.Queue.cancel_join_thread to_arduino_q.cancel_join_thread() to_parent_q.cancel_join_thread() # write_format: azimuth, altitude, R, G, & B intensity # read heartbeat: millis link = Serial( *SERVO_CONNECTION, read_timeout=7, error_pin=messageboard.GPIO_ERROR_ARDUINO_SERVO_CONNECTION, to_parent_q=to_parent_q, read_format='l', write_format='ffhhh', name='Servo') link.Open() last_flight = {} last_angles = (0, 0) flight, json_desc_dict, configuration, additional_attr = InitialMessageValues( to_arduino_q) next_read = 0 next_write = 0 now = GetNow(json_desc_dict, additional_attr) flight_present = False while not shutdown.value: if not to_arduino_q.empty(): flight, json_desc_dict, configuration, additional_attr = to_arduino_q.get( block=False) if 'test_servos' in configuration: messageboard.RemoveSetting(configuration, 'test_servos') <----SKIPPED LINES----> ('setting_off_time', 'H'), ('setting_delay', 'H'), ('last_plane', '?'), ('display_mode', 'H'), ('histogram_enabled', '?'), ('current_hist_type', 'H'), ('current_hist_history', 'H'), ('low_battery', '?')) write_config = ( ('setting_screen_enabled', '?'), # 1 bytes ('setting_max_distance', 'H'), # 2 bytes ('setting_max_altitude', 'L'), # 4 bytes ('setting_on_time', 'H'), # 2 bytes ('setting_off_time', 'H'), # 2 bytes ('setting_delay', 'H'), # 2 bytes ('line1', '9s'), # 9 bytes; 8 character plus terminator ('line2', '9s'), # 9 bytes; 8 character plus terminator ('line1_dec_mask', 'H'), # 2 bytes ('line2_dec_mask', 'H'), # 2 bytes ) #pylint: enable = bad-whitespace read_keys, unused_read_format_tuple, read_format_string = SplitFormat(read_config) write_keys, write_format_tuple, write_format_string = SplitFormat(write_config) values_d = {} low_batt = False to_parent_q.put(('pin', (messageboard.GPIO_ERROR_BATTERY_CHARGE, low_batt))) link = Serial( *REMOTE_CONNECTION, read_timeout=7, error_pin=messageboard.GPIO_ERROR_ARDUINO_REMOTE_CONNECTION, to_parent_q=to_parent_q, read_format=read_format_string, write_format=write_format_string, name='Remote') link.Open() display_mode = 2 #TODO: remove after switch integration flight, json_desc_dict, configuration, additional_attr = InitialMessageValues( to_arduino_q) next_read = 0 next_write = 0 while not shutdown.value: if not to_arduino_q.empty(): to_arduino_message = to_arduino_q.get(block=False) flight, json_desc_dict, configuration, additional_attr = to_arduino_message if 'test_remote' in configuration: messageboard.RemoveSetting(configuration, 'test_remote') def TestDisplayMode(m): <----SKIPPED LINES----> |