2
0

parse_bin.py 3.4 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  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.info(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.info(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('--source', help='Source file to parse')
  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.info(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. import configuration_pb2
  54. config = configuration_pb2.Config()
  55. with open(args.source, 'rb') as bin_file: # Open in binary mode
  56. data =bin_file.read()
  57. config.ParseFromString(data)
  58. logging.info(f'Parsed: {json.dumps(config)}')
  59. # for jsonfile in args.json:
  60. # output_file_base = os.path.join(args.target_dir, Path(jsonfile).stem+".bin")
  61. # logging.info(f'Converting JSON file {jsonfile} to binary format')
  62. # with open(jsonfile, 'r') as json_file:
  63. # json_data = json.load(json_file)
  64. # json_format.ParseDict(json_data, message)
  65. # binary_data = message.SerializeToString()
  66. # with open(output_file_base, 'wb') as bin_file:
  67. # bin_file.write(binary_data)
  68. # logging.info(f'Binary file written to {output_file_base}')
  69. if __name__ == '__main__':
  70. main()