01234567890123456789012345678901234567890123456789012345678901234567890123456789
756757758759760761762763764765766767768769770771772773774775 776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812 813814815816817818819820821822823824825826827828829830831832 866867868869870871872873874875876877878879880881882883884885886887888889890891892893 894 895896897898 899900901902903904905906 907908909910911912913914915916917918919920921922923924925926 936937938939940941942943944945946947948949950951952953954955 956957 958959960 961962963964965966967968969970971972973974975976977978979980981 982983984985986987988989990 991992993994995996997998 9991000100110021003100410051006100710081009101010111012101310141015101610171018 15901591159215931594159515961597159815991600160116021603160416051606160716081609 1610161116121613161416151616161716181619 |
<----SKIPPED LINES---->
for intensity in range(0, 101, 5):
b = intensity / 100
SendLEDTestMessage(*comms_args, *GammaRGB(rgb, gamma, brightness=b))
def ServoTestSweep(link, write_keys, write_format_tuple, altitude=45):
"""Sweep red laser around 360 degrees."""
for azimuth in range(0, 360, 10):
message_dict = GenerateServoMessage(
laser=LASER_RED, angles=(azimuth, altitude))
link.Write(DictToValueTuple(message_dict, write_keys, write_format_tuple))
time.sleep(WRITE_DELAY_TIME)
last_angles = (0, 0)
last_laser = LASER_OFF
last_led = LED_OFF
def GenerateServoMessage(
angles=None,
laser=None,
led=None,
reset=False):
"""Creates a dictionary of messages for servo arduino.
All values are optional; if not provided, the cached value last sent to the
Arduino is used.
"""
if angles: # if angles provided, update the cache
global last_angles
last_angles = angles
else: # if angles not provided, use last-provided angles
angles = last_angles
if laser:
global last_laser
last_laser = laser
else:
laser = last_laser
if led:
global last_led
last_led = led
else:
led = last_led
d = {}
d['azimuth'] = angles[0]
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, max_value=MAX_PWM):
"""Converts a desired brightness 0..max to a gamma-corrected PWM 0..max."""
# Based on https://hackaday.com/2016/08/23/
# rgb-leds-how-to-master-gamma-and-hue-for-perfect-brightness/
return ((c / max_value) ** gamma) * max_value
def PerceivedBrightness(rgb, gamma):
<----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()
#pylint: disable = bad-whitespace
write_config = (
('azimuth', 'f'), # 4 bytes
('altitude', 'f'), # 4 bytes
('laser_red', '?'), # 1 byte
('laser_green', '?'), # 1 byte
('laser_blue', '?'), # 1 byte
('led_red', 'H'), # 2 bytes
('led_green', 'H'), # 2 bytes
('led_blue', 'H'), # 2 bytes
('arduino_reset', '?'), # 1 byte
)
#pylint: enable = bad-whitespace
write_keys, write_format_tuple, write_format_string = SplitFormat(
write_config)
# write_format: azimuth, altitude, R, G, & B intensity
# read heartbeat: millis
link = Serial(
*SERVO_CONNECTION, read_timeout=60,
error_pin=messageboard.GPIO_ERROR_ARDUINO_SERVO_CONNECTION,
to_parent_q=to_parent_q,
read_format='l', write_format=write_format_string, name='Servo')
link.Open()
last_flight = {}
flight, json_desc_dict, configuration, additional_attr = InitialMessageValues(
to_arduino_q)
next_read = 0
next_write = 0
now = GetNow(json_desc_dict, additional_attr)
gamma = GetGamma(configuration)
brightness = 1.0
brightness_step = 0.1
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')
<----SKIPPED LINES---->
ServoTestDim(
link, write_keys, write_format_tuple,
HexColorToRGBTuple(configuration['led_color']), gamma)
gamma = GetGamma(configuration)
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))
now = GetNow(json_desc_dict, additional_attr)
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)
if brightness > 0:
brightness -= brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
message_dict = GenerateServoMessage(
laser=laser_rgb, angles=current_angles, led=rgb_tuple)
elif configuration['servo_mode'] == 'laser_only':
if brightness > 0:
brightness -= brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
message_dict = GenerateServoMessage(laser=LASER_OFF, led=rgb_tuple)
else:
if brightness < 1:
brightness += brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
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
# - to SFO / from SFO / other
<----SKIPPED LINES---->
SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
m, write_keys, write_format_tuple, link)
time.sleep(1)
TestDisplayMode(DISP_LAST_FLIGHT_NUMB_ORIG_DEST)
TestDisplayMode(DISP_LAST_FLIGHT_AZIMUTH_ELEVATION)
TestDisplayMode(DISP_FLIGHT_COUNT_LAST_SEEN)
TestDisplayMode(DISP_RADIO_RANGE)
if time.time() >= next_write:
next_write = SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
display_mode, write_keys, write_format_tuple, link)
if time.time() >= next_read:
bytes_read = []
values_t = link.Read(bytes_read=bytes_read)
values_d = dict(zip(read_keys, values_t))
if values_d.get('confirmed'):
Log(str(bytes_read) + '\n' + str(values_t) + '\n' + str(values_d))
display_mode, low_batt = ExecuteArduinoCommand(
values_d, configuration, display_mode, low_batt, to_parent_q, link)
next_read = time.time() + READ_DELAY_TIME_REMOTE
# send one final message telling Arduino to do a reset
SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
display_mode, write_keys, write_format_tuple, link, reset=True)
link.Close(SHUTDOWN_TEXT)
|
01234567890123456789012345678901234567890123456789012345678901234567890123456789
756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834 868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937 94794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055 1627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657 |
<----SKIPPED LINES---->
for intensity in range(0, 101, 5):
b = intensity / 100
SendLEDTestMessage(*comms_args, *GammaRGB(rgb, gamma, brightness=b))
def ServoTestSweep(link, write_keys, write_format_tuple, altitude=45):
"""Sweep red laser around 360 degrees."""
for azimuth in range(0, 360, 10):
message_dict = GenerateServoMessage(
laser=LASER_RED, angles=(azimuth, altitude))
link.Write(DictToValueTuple(message_dict, write_keys, write_format_tuple))
time.sleep(WRITE_DELAY_TIME)
last_angles = (0, 0)
last_laser = LASER_OFF
last_led = LED_OFF
def GenerateServoMessage(
mode_switch_ack=False,
angles=None,
laser=None,
led=None,
reset=False):
"""Creates a dictionary of messages for servo arduino.
All values are optional; if not provided, the cached value last sent to the
Arduino is used.
"""
if angles: # if angles provided, update the cache
global last_angles
last_angles = angles
else: # if angles not provided, use last-provided angles
angles = last_angles
if laser:
global last_laser
last_laser = laser
else:
laser = last_laser
if led:
global last_led
last_led = led
else:
led = last_led
d = {}
d['azimuth'] = angles[0]
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
d['mode_switch_ack'] = mode_switch_ack
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, max_value=MAX_PWM):
"""Converts a desired brightness 0..max to a gamma-corrected PWM 0..max."""
# Based on https://hackaday.com/2016/08/23/
# rgb-leds-how-to-master-gamma-and-hue-for-perfect-brightness/
return ((c / max_value) ** gamma) * max_value
def PerceivedBrightness(rgb, gamma):
<----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()
#pylint: disable = bad-whitespace
write_config = (
('azimuth', 'f'), # 4 bytes
('altitude', 'f'), # 4 bytes
('laser_red', '?'), # 1 byte
('laser_green', '?'), # 1 byte
('laser_blue', '?'), # 1 byte
('led_red', 'H'), # 2 bytes
('led_green', 'H'), # 2 bytes
('led_blue', 'H'), # 2 bytes
('mode_switch_ack', '?'), # 1 byte
('arduino_reset', '?'), # 1 byte
)
read_config = (
('mode_switch', '?'), # 1 byte
('millis', 'L'), # 4 bytes
)
#pylint: enable = bad-whitespace
write_keys, write_format_tuple, write_format_string = SplitFormat(
write_config)
read_keys, unused_read_format_tuple, read_format_string = SplitFormat(
read_config)
# write_format: azimuth, altitude, R, G, & B intensity
# read heartbeat: mode change & millis
link = Serial(
*SERVO_CONNECTION, read_timeout=60,
error_pin=messageboard.GPIO_ERROR_ARDUINO_SERVO_CONNECTION,
to_parent_q=to_parent_q,
read_format=read_format_string, write_format=write_format_string,
name='Servo')
link.Open()
last_flight = {}
flight, json_desc_dict, configuration, additional_attr = InitialMessageValues(
to_arduino_q)
next_read = 0
next_write = 0
now = GetNow(json_desc_dict, additional_attr)
gamma = GetGamma(configuration)
brightness = 1.0
brightness_step = 0.1
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')
<----SKIPPED LINES---->
ServoTestDim(
link, write_keys, write_format_tuple,
HexColorToRGBTuple(configuration['led_color']), gamma)
gamma = GetGamma(configuration)
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
mode_switch_ack = False
if time.time() >= next_read:
bytes_read = []
values_t = link.Read(bytes_read=bytes_read)
values_d = dict(zip(read_keys, values_t))
next_read = time.time() + READ_DELAY_TIME_SERVO
if bytes_read and VERBOSE:
Log('Read by Servo: %s' % str(values_d))
if values_d['mode_switch']:
# makes a copy so as to not modify underlying config; we don't want
# to modify underlying because otherwise the settings will bounce
# around (values read from disk -> new values set by arduino -> old
# values from disk -> new values from disk after Arduino update).
configuration = dict(configuration)
# LED only -> Laser only -> Both
sequence = ('led_only', 'laser_only', 'both')
old_name = configuration['servo_mode']
old_id = sequence.index(old_name)
new_id = (old_id + 1) % len(sequence)
new_name = sequence[new_id]
Log('Hemisphere mode updated from %s to %s' % (old_name, new_name))
configuration['servo_mode'] = new_name
settings_string = messageboard.BuildSettings(configuration)
to_parent_q.put(('update_configuration', (settings_string, )))
mode_switch_ack = True
now = GetNow(json_desc_dict, additional_attr)
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)
if brightness > 0:
brightness -= brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
message_dict = GenerateServoMessage(
mode_switch_ack=mode_switch_ack,
laser=laser_rgb, angles=current_angles, led=rgb_tuple)
elif configuration['servo_mode'] == 'laser_only':
if brightness > 0:
brightness -= brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
message_dict = GenerateServoMessage(
mode_switch_ack=mode_switch_ack, laser=LASER_OFF, led=rgb_tuple)
else:
if brightness < 1:
brightness += brightness_step
rgb_tuple = GammaRGB(
HexColorToRGBTuple(configuration['led_color']),
gamma, brightness=brightness)
message_dict = GenerateServoMessage(
mode_switch_ack=mode_switch_ack, 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
# - to SFO / from SFO / other
<----SKIPPED LINES---->
SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
m, write_keys, write_format_tuple, link)
time.sleep(1)
TestDisplayMode(DISP_LAST_FLIGHT_NUMB_ORIG_DEST)
TestDisplayMode(DISP_LAST_FLIGHT_AZIMUTH_ELEVATION)
TestDisplayMode(DISP_FLIGHT_COUNT_LAST_SEEN)
TestDisplayMode(DISP_RADIO_RANGE)
if time.time() >= next_write:
next_write = SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
display_mode, write_keys, write_format_tuple, link)
if time.time() >= next_read:
bytes_read = []
values_t = link.Read(bytes_read=bytes_read)
values_d = dict(zip(read_keys, values_t))
if values_d.get('confirmed'):
if VERBOSE:
Log(str(bytes_read) + '\n' + str(values_t) + '\n' + str(values_d))
display_mode, low_batt = ExecuteArduinoCommand(
values_d, configuration, display_mode, low_batt, to_parent_q, link)
next_read = time.time() + READ_DELAY_TIME_REMOTE
# send one final message telling Arduino to do a reset
SendRemoteMessage(
flight, json_desc_dict, configuration, additional_attr,
display_mode, write_keys, write_format_tuple, link, reset=True)
link.Close(SHUTDOWN_TEXT)
|