2
0

generate_bin.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import json
  2. import os
  3. import sys
  4. import argparse
  5. import importlib.util
  6. import logging
  7. from pathlib import Path
  8. from google.protobuf import json_format
  9. from google.protobuf.json_format import MessageToJson
  10. # Assuming this script is in the same directory as the generated Python files
  11. # script_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)),'generated/src')
  12. # sys.path.append(script_dir)
  13. # Configure logging
  14. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  15. def load_protobuf_module(source, includes):
  16. """Dynamically load a protobuf module given its file name."""
  17. for include in includes:
  18. sys.path.append(include)
  19. module_name = Path(source).stem
  20. module_file = module_name + '_pb2.py'
  21. module_location = Path(source).parent / module_file
  22. logging.debug(f'Loading module {module_file} from {module_location} with includes [{", ".join(includes)}]')
  23. spec = importlib.util.spec_from_file_location(name=module_name, location=str(module_location))
  24. if spec is not None:
  25. module = importlib.util.module_from_spec(spec)
  26. spec.loader.exec_module(module)
  27. logging.debug(f'Loaded protobuf module: {module_name}')
  28. return module
  29. else:
  30. logging.error(f'Failed to load module {module_file} from {module_location}')
  31. return None
  32. def protobuf_to_dict(message):
  33. """Convert a protobuf message to a dictionary."""
  34. # Convert the protobuf message to a JSON string
  35. json_str = MessageToJson(message, including_default_value_fields=True)
  36. # Parse the JSON string back into a dictionary
  37. return json.loads(json_str)
  38. def main():
  39. parser = argparse.ArgumentParser(description='Process protobuf and JSON files.')
  40. parser.add_argument('--proto_file', help='Name of the protobuf file (without extension)')
  41. parser.add_argument('--main_class', help='Main message class to process')
  42. parser.add_argument('--target_dir', help='Target directory for output files')
  43. parser.add_argument('--include', help='Directory where message python files can be found', default=None,action = 'append' )
  44. parser.add_argument('--json', help='Source JSON file(s)',action = 'append' )
  45. args = parser.parse_args()
  46. # Load the protobuf module
  47. logging.debug(f'Loading modules')
  48. proto_module = load_protobuf_module(args.proto_file, args.include)
  49. # Determine the main message class
  50. main_message_class = getattr(proto_module, args.main_class)
  51. message = main_message_class()
  52. proto_base_name = Path(args.proto_file).stem
  53. for jsonfile in args.json:
  54. output_file_base = os.path.join(args.target_dir, Path(jsonfile).stem+".bin")
  55. logging.debug(f'Converting JSON file {jsonfile} to binary format')
  56. with open(jsonfile, 'r') as json_file:
  57. json_data = json.load(json_file)
  58. json_format.ParseDict(json_data, message)
  59. binary_data = message.SerializeToString()
  60. with open(output_file_base, 'wb') as bin_file:
  61. bin_file.write(binary_data)
  62. logging.info(f'Binary file written to {output_file_base}')
  63. if __name__ == '__main__':
  64. main()