01234567890123456789012345678901234567890123456789012345678901234567890123456789
790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837 864865866867868869870871872873874875876877878879880881882883 884885886887888889890891892893894895896897898899900901902903 906907908909910911912913914915916917918919920921922923924925 926927928929930931932933934935936937938939940941942943944945946 | <----SKIPPED LINES----> d['altitude'] = angles[1] d['laser_red'] = laser[0] d['laser_green'] = laser[1] d['laser_blue'] = laser[2] d['led_red'] = led[0] d['led_green'] = led[1] d['led_blue'] = led[2] d['arduino_reset'] = reset return d def HexColorToRGBTuple(hex_color): """Converts i.e.: #329a43 to (50, 154, 67).""" r = hex_color[1:3] g = hex_color[3:5] b = hex_color[5:7] return (int(r, 16), int(g, 16), int(b, 16)) def Gamma(c): """Converts a desired brightness 0..255 to a gamma-corrected PWM 0..255.""" return round(((c / MAX_PWM) ** GAMMA) * MAX_PWM) def GammaRGB(rgb): """Gamma converts an RGB tuple.""" return Gamma(rgb[0]), Gamma(rgb[1]), Gamma(rgb[2]) 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() #pylint: disable = bad-whitespace write_config = ( <----SKIPPED LINES----> next_read = 0 next_write = 0 now = GetNow(json_desc_dict, additional_attr) while not shutdown.value: SetLoggingGlobals(configuration) if not to_arduino_q.empty(): flight, json_desc_dict, configuration, additional_attr = to_arduino_q.get( block=False) if 'test_servos_ordinal' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_ordinal') ServoTestOrdinal(link, write_keys, write_format_tuple) elif 'test_servos_sweep' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_sweep') ServoTestSweep(link, write_keys, write_format_tuple) elif 'test_servos_led' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_led') ServoTestLED(link, write_keys, write_format_tuple) new_flight = DifferentFlights(flight, last_flight) if new_flight: Log('Flight changed from %s to %s' % ( messageboard.DisplayFlightNumber(last_flight), messageboard.DisplayFlightNumber(flight) ), ser=link) # Turn off laser so line isn't traced while it moves to new position message_dict = GenerateServoMessage(laser=LASER_OFF) message_tuple = DictToValueTuple( message_dict, write_keys, write_format_tuple) link.Write(message_tuple) last_flight = flight if time.time() >= next_read: heartbeat = link.Read() # simple ack message sent by servos next_read = time.time() + READ_DELAY_TIME_SERVO if heartbeat and VERBOSE: Log('Heartbeat read by Servo: %s' % str(heartbeat)) <----SKIPPED LINES----> current_angles = AzimuthAltitude(flight, now) if time.time() > next_write: if (current_angles and current_angles[1] >= configuration['minimum_altitude_servo_tracking'] and configuration['servo_mode'] in ('laser_only', 'both')): if VERBOSE: Log('Flight #: %s current_angles: %s' % ( messageboard.DisplayFlightNumber(flight), str(current_angles))) laser_rgb = LaserRGBFlight(flight) message_dict = GenerateServoMessage( laser=laser_rgb, angles=current_angles, led=LED_OFF) elif configuration['servo_mode'] == 'laser_only': message_dict = GenerateServoMessage(laser=LASER_OFF, led=LED_OFF) else: rgb_tuple = GammaRGB(HexColorToRGBTuple(configuration['led_color'])) message_dict = GenerateServoMessage(laser=LASER_OFF, led=rgb_tuple) message_tuple = DictToValueTuple( message_dict, write_keys, write_format_tuple) link.Write(message_tuple) next_write = time.time() + WRITE_DELAY_TIME # One final write telling Arduino to do a software reset message_dict = GenerateServoMessage(laser=LASER_OFF, reset=True) message_tuple = DictToValueTuple(message_dict, write_keys, write_format_tuple) link.Write(message_tuple) link.Close(SHUTDOWN_TEXT) def LaserRGBFlight(flight): """Based on flight attributes, set the laser.""" # Possible assignment based on: # - ascending / descending / level <----SKIPPED LINES----> |
01234567890123456789012345678901234567890123456789012345678901234567890123456789
790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837 864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904 907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948 | <----SKIPPED LINES----> d['altitude'] = angles[1] d['laser_red'] = laser[0] d['laser_green'] = laser[1] d['laser_blue'] = laser[2] d['led_red'] = led[0] d['led_green'] = led[1] d['led_blue'] = led[2] d['arduino_reset'] = reset return d def HexColorToRGBTuple(hex_color): """Converts i.e.: #329a43 to (50, 154, 67).""" r = hex_color[1:3] g = hex_color[3:5] b = hex_color[5:7] return (int(r, 16), int(g, 16), int(b, 16)) def Gamma(c, gamma): """Converts a desired brightness 0..255 to a gamma-corrected PWM 0..255.""" return round(((c / MAX_PWM) ** gamma) * MAX_PWM) def GammaRGB(rgb, gamma): """Gamma converts an RGB tuple.""" return Gamma(rgb[0], gamma), Gamma(rgb[1], gamma), Gamma(rgb[2], gamma) 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() #pylint: disable = bad-whitespace write_config = ( <----SKIPPED LINES----> next_read = 0 next_write = 0 now = GetNow(json_desc_dict, additional_attr) while not shutdown.value: SetLoggingGlobals(configuration) if not to_arduino_q.empty(): flight, json_desc_dict, configuration, additional_attr = to_arduino_q.get( block=False) if 'test_servos_ordinal' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_ordinal') ServoTestOrdinal(link, write_keys, write_format_tuple) elif 'test_servos_sweep' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_sweep') ServoTestSweep(link, write_keys, write_format_tuple) elif 'test_servos_led' in configuration: messageboard.RemoveSetting(configuration, 'test_servos_led') ServoTestLED(link, write_keys, write_format_tuple) gamma = configuration['servo_rgb_gamma'] new_flight = DifferentFlights(flight, last_flight) if new_flight: Log('Flight changed from %s to %s' % ( messageboard.DisplayFlightNumber(last_flight), messageboard.DisplayFlightNumber(flight) ), ser=link) # Turn off laser so line isn't traced while it moves to new position message_dict = GenerateServoMessage(laser=LASER_OFF) message_tuple = DictToValueTuple( message_dict, write_keys, write_format_tuple) link.Write(message_tuple) last_flight = flight if time.time() >= next_read: heartbeat = link.Read() # simple ack message sent by servos next_read = time.time() + READ_DELAY_TIME_SERVO if heartbeat and VERBOSE: Log('Heartbeat read by Servo: %s' % str(heartbeat)) <----SKIPPED LINES----> current_angles = AzimuthAltitude(flight, now) if time.time() > next_write: if (current_angles and current_angles[1] >= configuration['minimum_altitude_servo_tracking'] and configuration['servo_mode'] in ('laser_only', 'both')): if VERBOSE: Log('Flight #: %s current_angles: %s' % ( messageboard.DisplayFlightNumber(flight), str(current_angles))) laser_rgb = LaserRGBFlight(flight) message_dict = GenerateServoMessage( laser=laser_rgb, angles=current_angles, led=LED_OFF) elif configuration['servo_mode'] == 'laser_only': message_dict = GenerateServoMessage(laser=LASER_OFF, led=LED_OFF) else: rgb_tuple = GammaRGB( HexColorToRGBTuple(configuration['led_color']), gamma) message_dict = GenerateServoMessage(laser=LASER_OFF, led=rgb_tuple) message_tuple = DictToValueTuple( message_dict, write_keys, write_format_tuple) link.Write(message_tuple) next_write = time.time() + WRITE_DELAY_TIME # One final write telling Arduino to do a software reset message_dict = GenerateServoMessage(laser=LASER_OFF, reset=True) message_tuple = DictToValueTuple(message_dict, write_keys, write_format_tuple) link.Write(message_tuple) link.Close(SHUTDOWN_TEXT) def LaserRGBFlight(flight): """Based on flight attributes, set the laser.""" # Possible assignment based on: # - ascending / descending / level <----SKIPPED LINES----> |