Background

Eventually, we’ll have many types of telemetry data. To make it easier to add new telemetry, we can look into writing a code generator. See the important information for a snippet on how to implement it.

Requirements

• Take the example in the important information section and convert it to use TOML files as that's what we use for our configuration code generator files
• Take an existing telemetry and convert it to use the code generator script
• Validate to make sure the output is correct

Important Information

Message Format JSON

• Here’s an example JSON file that defines the format of a few telemetry messages.

  {
      "messages": [
          {
              "id": 1,
              "name": "orbit_data",
              "signals": [
                  {
                      "name": "altitude",
                      "bit_width": 20,
                      "min_value": 0,
                      "max_value": 1000,
                      "value_type": "continuous"
                  },
                  {
                      "name": "velocity",
                      "bit_width": 21,
                      "min_value": 0,
                      "max_value": 30000,
                      "value_type": "continuous"
                  },
                  {
                      "name": "orbit_type",
                      "bit_width": 3,
                      "min_value": 0,
                      "max_value": 7,
                      "value_type": "discrete"
                  },
                  {
                      "name": "orbit_period",
                      "bit_width": 25,
                      "min_value": 0,
                      "max_value": 20000,
                      "value_type": "continuous"
                  }
              ]
          },
          {
              "id": 2,
              "name": "system_health",
              "signals": [
                  {
                      "name": "cpu_temp",
                      "bit_width": 15,
                      "min_value": -40,
                      "max_value": 85,
                      "value_type": "continuous"
                  },
                  {
                      "name": "battery_level",
                      "bit_width": 7,
                      "min_value": 0,
                      "max_value": 100,
                      "value_type": "discrete"
                  }
              ]
          },
          {
              "id": 3,
              "name": "comms_status",
              "signals": [
                  {
                      "name": "signal_strength",
                      "bit_width": 6,
                      "min_value": 0,
                      "max_value": 100,
                      "value_type": "continuous"
                  },
                  {
                      "name": "data_rate",
                      "bit_width": 24,
                      "min_value": 0,
                      "max_value": 10000,
                      "value_type": "continuous"
                  }
              ]
          }
      ]
  }

Code Generator Script

• Here’s a proof-of-concept code generator.

  import json
  
  def generate_c_macros(messages):
      c_code = []
  
      for message in messages:
          c_code.append(f"// {message['name']}\n")
          c_code.append(f"#define telemetry_{message['name']}_msg_id {message['id']}\n")
                  
          for signal in message['signals']:
              c_code.append(f"#define telemetry_{message['name']}_{signal['name']}_max {signal['max_value']}\n")
              c_code.append(f"#define telemetry_{message['name']}_{signal['name']}_min {signal['min_value']}\n")       
              
          c_code.append("\n")         
          
      return ''.join(c_code)
  
  def generate_c_code_msg_defs(messages):
      c_code = []
  
      for message in messages:
          c_code.append(f"typedef struct {{\n")
          for signal in message['signals']:
              if signal['bit_width'] <= 8:
                  datatype = "uint8_t"
              elif signal['bit_width'] <= 16:
                  datatype = "uint16_t"
              elif signal['bit_width'] <= 32:
                  datatype = "uint32_t"
              elif signal['bit_width'] <= 64:
                  datatype = "uint64_t"
              else:
                  raise Exception("Signal bit width is too large")
              c_code.append(f"    {datatype} {signal['name']}; // {signal['bit_width']} bits\n")
          c_code.append(f"}} telemetry_{message['name']}_msg_t;\n\n")
      
      return ''.join(c_code)
  
  def generate_c_code_msg_mailboxes(messages):
      c_code = []
  
      for message in messages:
          c_code.append(f"static telemetry_{message['name']}_msg_t {message['name']}_mailbox = {{0}};\n")
      
      c_code.append("\n")
      
      return ''.join(c_code)
  
  def generate_c_set_get_functions(messages):
      c_code = []
  
      for message in messages:
          for signal in message['signals']:
              max_value = 2 ** signal['bit_width'] - 1
              range_value = signal['max_value'] - signal['min_value']
  
              # Determine the smallest integer type for the signal
              if signal['bit_width'] <= 8:
                  datatype = "uint8_t"
              elif signal['bit_width'] <= 16:
                  datatype = "uint16_t"
              elif signal['bit_width'] <= 32:
                  datatype = "uint32_t"
              else:
                  datatype = "uint64_t"
  
              if signal['value_type'] == 'continuous':
                  set_func = (
                      f"void telemetry_set_{message['name']}_{signal['name']}(float value) {{\n"
                      f"    value = (value - {signal['min_value']}) / {range_value} * {hex(max_value)}; \n"
                      f"    {message['name']}_mailbox.{signal['name']} = ({datatype})(value + 0.5);\n"
                      f"}}\n\n"
                  )
                  get_func = (
                      f"float telemetry_get_{message['name']}_{signal['name']}(void) {{\n"
                      f"    {datatype} value = {message['name']}_mailbox.{signal['name']};\n"
                      f"    return ((float)value / {hex(max_value)} * {range_value}) + {signal['min_value']};\n"
                      f"}}\n\n"
                  )
              else:  # discrete
                  set_func = (
                      f"void telemetry_set_{message['name']}_{signal['name']}({datatype} value) {{\n"
                      f"    {message['name']}_mailbox.{signal['name']} = value & {hex(max_value)};\n"
                      f"}}\n\n"
                  )
                  get_func = (
                      f"{datatype} telemetry_get_{message['name']}_{signal['name']}(void) {{\n"
                      f"    return {message['name']}_mailbox.{signal['name']};\n"
                      f"}}\n\n"
                  )
      
              c_code.append(set_func)
              c_code.append(get_func)
  
      return ''.join(c_code)
  
  def generate_c_serialization_deserialization_functions(messages):
      c_code = []
  
      for message in messages:
          total_bits = sum(signal['bit_width'] for signal in message['signals'])
          total_bytes = (total_bits + 7) // 8  # Total bytes for serialized message, with padding
  
          # Serialization function
          serialize_func = (
              f"void serialize_{message['name']}(uint8_t* buffer) {{\n"
              f"    uint64_t bit_stream = 0U;\n"
              f"    uint32_t bit_offset = 0U;\n"
          )
  
          # Loop through each signal for serialization
          for signal in message['signals']:
              serialize_func += (
                  f"    bit_stream |= ((uint64_t){message['name']}_mailbox.{signal['name']} & {2**signal['bit_width'] - 1}) << bit_offset;\n"
                  f"    bit_offset += {signal['bit_width']};\n"
              )
  
          # Add padding bits (if necessary) and convert bit stream to byte array
          serialize_func += (
              f"    for (int i = 0; i < {total_bytes}; ++i) {{\n"
              f"        buffer[i] = (bit_stream >> (8 * i)) & 0xFF;\n"
              f"    }}\n"
              f"}}\n\n"
          )
          c_code.append(serialize_func)
  
          # Deserialization function
          deserialize_func = (
              f"void deserialize_{message['name']}(uint8_t* buffer) {{\n"
              f"    uint64_t bit_stream = 0U;\n"
              f"    uint32_t bit_offset = 0U;\n"
              f"    for (int i = 0; i < {total_bytes}; ++i) {{\n"
              f"        bit_stream |= ((uint64_t)buffer[i]) << (8 * i);\n"
              f"    }}\n"
          )
  
          # Loop through each signal for deserialization
          bit_offset = 0
          for signal in message['signals']:
              deserialize_func += (
                  f"    {message['name']}_mailbox.{signal['name']} = (bit_stream >> {bit_offset}) & {hex(2**signal['bit_width'] - 1)};\n"
                  f"    bit_offset += {signal['bit_width']};\n"
              )    
          deserialize_func += f"}}\n\n"
          c_code.append(deserialize_func)
      
      return ''.join(c_code)
  
  def generate_c_code_with_struct_and_macros(data):
      """Generate C code with advanced serialization, mailbox struct, and min/max macros."""
      c_code = []
  
      c_code.append(generate_c_macros(data['messages']))
      c_code.append(generate_c_code_msg_mailboxes(data['messages']))
      c_code.append(generate_c_code_msg_defs(data['messages']))
      c_code.append(generate_c_set_get_functions(data['messages']))
      c_code.append(generate_c_serialization_deserialization_functions(data['messages']))
  
      return ''.join(c_code)
  
  # Generate the C code with advanced serialization for the non-standard JSON data
  c_module_code_advanced_serialization = generate_c_code_with_struct_and_macros(json.load(open('telem.json')))
  
  with open("telem.h", "w") as f:
      f.write("#pragma once\n\n")
      f.write("#include <stdint.h>\n\n")
      f.write(c_module_code_advanced_serialization)

Example Output

• Here’s the output of the code generator

  #pragma once
  
  #include <stdint.h>
  
  // orbit_data
  #define telemetry_orbit_data_msg_id 1
  #define telemetry_orbit_data_altitude_max 1000
  #define telemetry_orbit_data_altitude_min 0
  #define telemetry_orbit_data_velocity_max 30000
  #define telemetry_orbit_data_velocity_min 0
  #define telemetry_orbit_data_orbit_type_max 7
  #define telemetry_orbit_data_orbit_type_min 0
  #define telemetry_orbit_data_orbit_period_max 20000
  #define telemetry_orbit_data_orbit_period_min 0
  
  // system_health
  #define telemetry_system_health_msg_id 2
  #define telemetry_system_health_cpu_temp_max 85
  #define telemetry_system_health_cpu_temp_min -40
  #define telemetry_system_health_battery_level_max 100
  #define telemetry_system_health_battery_level_min 0
  
  // comms_status
  #define telemetry_comms_status_msg_id 3
  #define telemetry_comms_status_signal_strength_max 100
  #define telemetry_comms_status_signal_strength_min 0
  #define telemetry_comms_status_data_rate_max 10000
  #define telemetry_comms_status_data_rate_min 0
  
  static telemetry_orbit_data_msg_t orbit_data_mailbox = {0};
  static telemetry_system_health_msg_t system_health_mailbox = {0};
  static telemetry_comms_status_msg_t comms_status_mailbox = {0};
  
  typedef struct {
      uint32_t altitude; // 20 bits
      uint32_t velocity; // 21 bits
      uint8_t orbit_type; // 3 bits
      uint32_t orbit_period; // 25 bits
  } telemetry_orbit_data_msg_t;
  
  typedef struct {
      uint16_t cpu_temp; // 15 bits
      uint8_t battery_level; // 7 bits
  } telemetry_system_health_msg_t;
  
  typedef struct {
      uint8_t signal_strength; // 6 bits
      uint32_t data_rate; // 24 bits
  } telemetry_comms_status_msg_t;
  
  void telemetry_set_orbit_data_altitude(float value) {
      value = (value - 0) / 1000 * 0xfffff; 
      orbit_data_mailbox.altitude = (uint32_t)(value + 0.5);
  }
  
  float telemetry_get_orbit_data_altitude(void) {
      uint32_t value = orbit_data_mailbox.altitude;
      return ((float)value / 0xfffff * 1000) + 0;
  }
  
  void telemetry_set_orbit_data_velocity(float value) {
      value = (value - 0) / 30000 * 0x1fffff; 
      orbit_data_mailbox.velocity = (uint32_t)(value + 0.5);
  }
  
  float telemetry_get_orbit_data_velocity(void) {
      uint32_t value = orbit_data_mailbox.velocity;
      return ((float)value / 0x1fffff * 30000) + 0;
  }
  
  void telemetry_set_orbit_data_orbit_type(uint8_t value) {
      orbit_data_mailbox.orbit_type = value & 0x7;
  }
  
  uint8_t telemetry_get_orbit_data_orbit_type(void) {
      return orbit_data_mailbox.orbit_type;
  }
  
  void telemetry_set_orbit_data_orbit_period(float value) {
      value = (value - 0) / 20000 * 0x1ffffff; 
      orbit_data_mailbox.orbit_period = (uint32_t)(value + 0.5);
  }
  
  float telemetry_get_orbit_data_orbit_period(void) {
      uint32_t value = orbit_data_mailbox.orbit_period;
      return ((float)value / 0x1ffffff * 20000) + 0;
  }
  
  void telemetry_set_system_health_cpu_temp(float value) {
      value = (value - -40) / 125 * 0x7fff; 
      system_health_mailbox.cpu_temp = (uint16_t)(value + 0.5);
  }
  
  float telemetry_get_system_health_cpu_temp(void) {
      uint16_t value = system_health_mailbox.cpu_temp;
      return ((float)value / 0x7fff * 125) + -40;
  }
  
  void telemetry_set_system_health_battery_level(uint8_t value) {
      system_health_mailbox.battery_level = value & 0x7f;
  }
  
  uint8_t telemetry_get_system_health_battery_level(void) {
      return system_health_mailbox.battery_level;
  }
  
  void telemetry_set_comms_status_signal_strength(float value) {
      value = (value - 0) / 100 * 0x3f; 
      comms_status_mailbox.signal_strength = (uint8_t)(value + 0.5);
  }
  
  float telemetry_get_comms_status_signal_strength(void) {
      uint8_t value = comms_status_mailbox.signal_strength;
      return ((float)value / 0x3f * 100) + 0;
  }
  
  void telemetry_set_comms_status_data_rate(float value) {
      value = (value - 0) / 10000 * 0xffffff; 
      comms_status_mailbox.data_rate = (uint32_t)(value + 0.5);
  }
  
  float telemetry_get_comms_status_data_rate(void) {
      uint32_t value = comms_status_mailbox.data_rate;
      return ((float)value / 0xffffff * 10000) + 0;
  }
  
  void serialize_orbit_data(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      bit_stream |= ((uint64_t)orbit_data_mailbox.altitude & 1048575) << bit_offset;
      bit_offset += 20;
      bit_stream |= ((uint64_t)orbit_data_mailbox.velocity & 2097151) << bit_offset;
      bit_offset += 21;
      bit_stream |= ((uint64_t)orbit_data_mailbox.orbit_type & 7) << bit_offset;
      bit_offset += 3;
      bit_stream |= ((uint64_t)orbit_data_mailbox.orbit_period & 33554431) << bit_offset;
      bit_offset += 25;
      for (int i = 0; i < 9; ++i) {
          buffer[i] = (bit_stream >> (8 * i)) & 0xFF;
      }
  }
  
  void deserialize_orbit_data(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      for (int i = 0; i < 9; ++i) {
          bit_stream |= ((uint64_t)buffer[i]) << (8 * i);
      }
      orbit_data_mailbox.altitude = (bit_stream >> 0) & 0xfffff;
      bit_offset += 20;
      orbit_data_mailbox.velocity = (bit_stream >> 0) & 0x1fffff;
      bit_offset += 21;
      orbit_data_mailbox.orbit_type = (bit_stream >> 0) & 0x7;
      bit_offset += 3;
      orbit_data_mailbox.orbit_period = (bit_stream >> 0) & 0x1ffffff;
      bit_offset += 25;
  }
  
  void serialize_system_health(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      bit_stream |= ((uint64_t)system_health_mailbox.cpu_temp & 32767) << bit_offset;
      bit_offset += 15;
      bit_stream |= ((uint64_t)system_health_mailbox.battery_level & 127) << bit_offset;
      bit_offset += 7;
      for (int i = 0; i < 3; ++i) {
          buffer[i] = (bit_stream >> (8 * i)) & 0xFF;
      }
  }
  
  void deserialize_system_health(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      for (int i = 0; i < 3; ++i) {
          bit_stream |= ((uint64_t)buffer[i]) << (8 * i);
      }
      system_health_mailbox.cpu_temp = (bit_stream >> 0) & 0x7fff;
      bit_offset += 15;
      system_health_mailbox.battery_level = (bit_stream >> 0) & 0x7f;
      bit_offset += 7;
  }
  
  void serialize_comms_status(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      bit_stream |= ((uint64_t)comms_status_mailbox.signal_strength & 63) << bit_offset;
      bit_offset += 6;
      bit_stream |= ((uint64_t)comms_status_mailbox.data_rate & 16777215) << bit_offset;
      bit_offset += 24;
      for (int i = 0; i < 4; ++i) {
          buffer[i] = (bit_stream >> (8 * i)) & 0xFF;
      }
  }
  
  void deserialize_comms_status(uint8_t* buffer) {
      uint64_t bit_stream = 0U;
      uint32_t bit_offset = 0U;
      for (int i = 0; i < 4; ++i) {
          bit_stream |= ((uint64_t)buffer[i]) << (8 * i);
      }
      comms_status_mailbox.signal_strength = (bit_stream >> 0) & 0x3f;
      bit_offset += 6;
      comms_status_mailbox.data_rate = (bit_stream >> 0) & 0xffffff;
      bit_offset += 24;
  }