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---->
|