01234567890123456789012345678901234567890123456789012345678901234567890123456789
22542255225622572258225922602261226222632264226522662267226822692270227122722273 22742275227622772278227922802281228222832284228522862287228822892290229122922293 60806081608260836084608560866087608860896090609160926093609460956096609760986099 610061016102610361046105610661076108610961106111611261136114611561166117611861196120612161226123612461256126612761286129613061316132613361346135613661376138613961406141614261436144614561466147614861496150615161526153615461556156615761586159616061616162 | <----SKIPPED LINES----> destination_length = len(destination_iata) origin_length = max_pair_length - destination_length elif destination_iata == origin_iata: origin_length = len(origin_iata) destination_length = max_pair_length - origin_length else: destination_length = len(destination_iata) origin_length = len(origin_iata) if (origin_iata == KEY_NOT_PRESENT_STRING and destination_iata == KEY_NOT_PRESENT_STRING): origin_destination_pair = KEY_NOT_PRESENT_STRING else: origin_destination_pair = ('%s-%s' % ( origin_friendly[:origin_length], destination_friendly[:destination_length])) return origin_destination_pair def DisplayDepartureTimes(flight): """Generates displayable fields about flight times and delay. Attempts to first find matching "pairs" of flight departure time details (departure vs. takeoff) in the belief that aligned nomenclature in the source data reflects an aligned concept of time where a flight delay can be best calculated. Without a matching pair (or if perhaps no departure time information is provided), then a delay cannot be calculated at all. Args: flight: dictionary with key-value attributes about the flight. Returns: Dictionary with the following keys: - departure_timestamp: taken from one of potentially four timestamps indicating departure - departure_time_text: departure time formatted to HH:MM string - calculable_delay: boolean indicating whether sufficient data available to calc delay - delay_seconds: integer number of seconds of delay <----SKIPPED LINES----> else: histogram_type, histogram_history = args message_queue.extend(MessageboardHistograms( flights, histogram_type, histogram_history, '_1', False)) elif command == 'update_configuration': updated_settings = args[0] Log('Updated settings received from arduino: %s' % updated_settings) WriteFile(CONFIG_FILE, updated_settings) else: Log('Improper command from arduinos: %s / %s' % (command, args)) return message_queue, next_message_time def PublishMessage( s, subscription_id='12fd73cd-75ef-4cae-bbbf-29b2678692c1', key='c5f62d44-e30d-4c43-a43e-d4f65f4eb399', secret='b00aeb24-72f3-467c-aad2-82ba5e5266ca', timeout=3): """Publishes a text string to a Vestaboard. The message is pushed to the vestaboard splitflap display by way of its web services; see https://docs.vestaboard.com/introduction for more details. Args: s: String to publish. subscription_id: string subscription id from Vestaboard. key: string key from Vestaboard. secret: string secret from Vestaboard. timeout: Max duration in seconds that we should wait to establish a connection. """ error_code = False # See https://docs.vestaboard.com/characters: any chars needing to be replaced special_characters = ((u'\u00b0', '{62}'),) # degree symbol '°' for special_character in special_characters: s = s.replace(*(special_character)) curl = pycurl.Curl() # See https://stackoverflow.com/questions/31826814/ # curl-post-request-into-pycurl-code # Set URL value curl.setopt( pycurl.URL, 'https://platform.vestaboard.com/subscriptions/%s/message' % subscription_id) curl.setopt(pycurl.HTTPHEADER, [ 'X-Vestaboard-Api-Key:%s' % key, 'X-Vestaboard-Api-Secret:%s' % secret]) curl.setopt(pycurl.TIMEOUT_MS, timeout*1000) curl.setopt(pycurl.POST, 1) curl.setopt(pycurl.WRITEFUNCTION, lambda x: None) # to keep stdout clean # preparing body the way pycurl.READDATA wants it body_as_dict = {'text': s} body_as_json_string = json.dumps(body_as_dict) # dict to json body_as_file_object = io.StringIO(body_as_json_string) # prepare and send. See also: pycurl.READFUNCTION to pass function instead curl.setopt(pycurl.READDATA, body_as_file_object) curl.setopt(pycurl.POSTFIELDSIZE, len(body_as_json_string)) failure_message = '' try: curl.perform() except pycurl.error as e: failure_message = 'curl.perform() failed with message %s' % e Log('curl.perform() failed with message %s' % e) error_code = True else: # you may want to check HTTP response code, e.g. status_code = curl.getinfo(pycurl.RESPONSE_CODE) if status_code != 200: Log('Server returned HTTP status code %d for message %s' % ( status_code, s)) error_code = True <----SKIPPED LINES----> |
01234567890123456789012345678901234567890123456789012345678901234567890123456789
225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313 61006101610261036104610561066107610861096110611161126113611461156116611761186119612061216122612361246125612661276128612961306131613261336134613561366137613861396140614161426143614461456146614761486149615061516152615361546155615661576158615961606161616261636164616561666167616861696170617161726173617461756176617761786179618061816182618361846185618661876188618961906191619261936194619561966197619861996200620162026203620462056206620762086209621062116212621362146215621662176218621962206221622262236224622562266227622862296230623162326233623462356236623762386239 62406241624262436244624562466247624862496250625162526253625462556256625762586259626062616262626362646265626662676268626962706271627262736274627562766277 | <----SKIPPED LINES----> destination_length = len(destination_iata) origin_length = max_pair_length - destination_length elif destination_iata == origin_iata: origin_length = len(origin_iata) destination_length = max_pair_length - origin_length else: destination_length = len(destination_iata) origin_length = len(origin_iata) if (origin_iata == KEY_NOT_PRESENT_STRING and destination_iata == KEY_NOT_PRESENT_STRING): origin_destination_pair = KEY_NOT_PRESENT_STRING else: origin_destination_pair = ('%s-%s' % ( origin_friendly[:origin_length], destination_friendly[:destination_length])) return origin_destination_pair def DisplayOriginDestinationPairShort(flight): """Generates displayble origin-dest airport code. Generates airport code pairs if both origin & destination are known, and None otherwise (i.e.: SFO-SJC). Args: flight: dictionary with key-value attributes about the flight. Returns: String or None as described. """ origin_iata = DisplayOriginIata(flight) destination_iata = DisplayDestinationIata(flight) if KEY_NOT_PRESENT_STRING in (origin_iata, destination_iata): return None return '%s-%s' % (origin_iata, destination_iata) def DisplayDepartureTimes(flight): """Generates displayable fields about flight times and delay. Attempts to first find matching "pairs" of flight departure time details (departure vs. takeoff) in the belief that aligned nomenclature in the source data reflects an aligned concept of time where a flight delay can be best calculated. Without a matching pair (or if perhaps no departure time information is provided), then a delay cannot be calculated at all. Args: flight: dictionary with key-value attributes about the flight. Returns: Dictionary with the following keys: - departure_timestamp: taken from one of potentially four timestamps indicating departure - departure_time_text: departure time formatted to HH:MM string - calculable_delay: boolean indicating whether sufficient data available to calc delay - delay_seconds: integer number of seconds of delay <----SKIPPED LINES----> else: histogram_type, histogram_history = args message_queue.extend(MessageboardHistograms( flights, histogram_type, histogram_history, '_1', False)) elif command == 'update_configuration': updated_settings = args[0] Log('Updated settings received from arduino: %s' % updated_settings) WriteFile(CONFIG_FILE, updated_settings) else: Log('Improper command from arduinos: %s / %s' % (command, args)) return message_queue, next_message_time def StringToCharArray(s): '''Converts a string to a character array so as to preserve spacing et al. If passed a 'text' message, the Vestaboard a) removes duplicate spaces and b) left justifies. This messes up messages with centered headers or tabular columns. Thus, the 'character' mode must be used where a 6x22 2-d array of character codes is passed as the message. Character codes are specified at https://docs.vestaboard.com/characters, so an A, for instance, needs to be translated to code 1. Additionally, escaped characters (such as colors or symbols), which are specified in text such as {64} for orange, for example, need to be passed simply as 64. Args: s: a string representing 132 characters to be displayed, albeit it may be longer than 132 characters due to embedded escaped characters. Returns: A 2-d array of character codes, i.e.: [[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]] where each 0 is replaced by an integer 0,...,69. ''' # For each element in the string, # if it is an '{', find the closing '}' and collapse into one element # if it is not an '{' and we are not in an escaped string, translate translation = { ' ': 0, 'A': 1, 'B': 2, 'C': 3, 'D': 4, 'E': 5, 'F': 6, 'G': 7, 'H': 8, 'I': 9, 'J': 10, 'K': 11, 'L': 12, 'M': 13, 'N': 14, 'O': 15, 'P': 16, 'Q': 17, 'R': 18, 'S': 19, 'T': 20, 'U': 21, 'V': 22, 'W': 23, 'X': 24, 'Y': 25, 'Z': 26, '1': 27, '2': 28, '3': 29, '4': 30, '5': 31, '6': 32, '7': 33, '8': 34, '9': 35, '0': 36, '!': 37, '@': 38, '#': 39, '$': 40, '(': 41, ')': 42, '-': 44, '+': 46, '&': 47, '=': 48, ';': 49, ':': 50, "'": 52, '"': 53, '%': 54, ',': 55, '.': 56, '/': 59, '?': 60, '°': 62} # valid codes are those enumerated above plus the seven colors in 63, ..., 69 valid_codes = list(translation.values()) valid_codes.extend([63, 64, 65, 66, 67, 68, 69]) pointer = 0 message_array = [] while pointer < len(s): if s[pointer] == '{': end_escape = s.find('}', pointer) if end_escape == -1: Log('Escaped sequence missing closing curly brace: "%s"' % s) escaped_sequence = s[pointer+1 : end_escape] try: escaped_value = int(escaped_sequence) except ValueError: Log('Escaped sequence "%s" is not a number in message "%s"' % (escaped_sequence, s)) if escaped_value not in valid_codes: Log('Escaped sequence "%d" is not a valid escape code in message "%s"' % (escaped_value, s)) else: message_array.append(escaped_value) if end_escape == -1: pointer = len(s) else: pointer = end_escape + 1 else: message_array.append(translation[s[pointer].upper()]) pointer += 1 expected_characters = SPLITFLAP_CHARS_PER_LINE * SPLITFLAP_LINE_COUNT missing_characters = max(0, expected_characters - len(message_array)) if missing_characters: for unused_n in range(missing_characters): message_array.append(0) extra_characters = max(0, len(message_array) - expected_characters) if extra_characters: Log('Message is too long at %d characters (max %d characters)' % (len(message_array), expected_characters)) message_array = message_array[:expected_characters] message_2d_array = [] for line_num in range(SPLITFLAP_LINE_COUNT): message_2d_array.append(message_array[ line_num * SPLITFLAP_CHARS_PER_LINE : (line_num + 1)*SPLITFLAP_CHARS_PER_LINE]) return message_2d_array def PublishMessage( s, subscription_id='12fd73cd-75ef-4cae-bbbf-29b2678692c1', key='c5f62d44-e30d-4c43-a43e-d4f65f4eb399', secret='b00aeb24-72f3-467c-aad2-82ba5e5266ca', timeout=3): """Publishes a text string to a Vestaboard. The message is pushed to the vestaboard splitflap display by way of its web services; see https://docs.vestaboard.com/introduction for more details. Args: s: String to publish. subscription_id: string subscription id from Vestaboard. key: string key from Vestaboard. secret: string secret from Vestaboard. timeout: Max duration in seconds that we should wait to establish a connection. """ error_code = False curl = pycurl.Curl() # See https://stackoverflow.com/questions/31826814/ # curl-post-request-into-pycurl-code # Set URL value curl.setopt( pycurl.URL, 'https://platform.vestaboard.com/subscriptions/%s/message' % subscription_id) curl.setopt(pycurl.HTTPHEADER, [ 'X-Vestaboard-Api-Key:%s' % key, 'X-Vestaboard-Api-Secret:%s' % secret]) curl.setopt(pycurl.TIMEOUT_MS, timeout*1000) curl.setopt(pycurl.POST, 1) curl.setopt(pycurl.WRITEFUNCTION, lambda x: None) # to keep stdout clean # preparing body the way pycurl.READDATA wants it body_as_dict = {'characters': StringToCharArray(s)} body_as_json_string = json.dumps(body_as_dict) # dict to json body_as_file_object = io.StringIO(body_as_json_string) # prepare and send. See also: pycurl.READFUNCTION to pass function instead curl.setopt(pycurl.READDATA, body_as_file_object) curl.setopt(pycurl.POSTFIELDSIZE, len(body_as_json_string)) failure_message = '' try: curl.perform() except pycurl.error as e: failure_message = 'curl.perform() failed with message %s' % e Log('curl.perform() failed with message %s' % e) error_code = True else: # you may want to check HTTP response code, e.g. status_code = curl.getinfo(pycurl.RESPONSE_CODE) if status_code != 200: Log('Server returned HTTP status code %d for message %s' % ( status_code, s)) error_code = True <----SKIPPED LINES----> |