1. import os
  2. import subprocess
  3. import logging
  4. import json
  5. from datetime import datetime
  7. # Configure logging
  8. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  10. def bootstrap_user_record(user_id, experiment_config_path, script_path):
  11. """
  12. Bootstraps a user record script for an experiment.
  14. Args:
  15. user_id (str): The ID of the user.
  16. experiment_config_path (str): Path to the experiment configuration file (JSON).
  17. script_path (str): Path to the script to be executed.
  19. Returns:
  20. bool: True if the script ran successfully, False otherwise.
  21. """
  23. try:
  24. # Load experiment configuration
  25. with open(experiment_config_path, 'r') as f:
  26. experiment_config = json.load(f)
  28. # Prepare input for the script
  29. input_data = {"user_id": user_id, **experiment_config.get("input_parameters", {})}
  31. # Construct the command to execute the script
  32. command = [script_path] + list(input_data.values())
  34. logging.info(f"Executing script with command: {' '.join(command)}")
  36. # Execute the script
  37. process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
  38. stdout, stderr = process.communicate()
  40. # Check the return code
  41. if process.returncode == 0:
  42. logging.info(f"Script executed successfully for user {user_id}.")
  43. logging.debug(f"Script output:\n{stdout}")
  44. return True
  45. else:
  46. logging.error(f"Script failed for user {user_id}. Return code: {process.returncode}")
  47. logging.error(f"Script error output:\n{stderr}")
  48. return False
  50. except FileNotFoundError:
  51. logging.error(f"Script not found at {script_path}")
  52. return False
  53. except json.JSONDecodeError:
  54. logging.error(f"Error decoding JSON from {experiment_config_path}")
  55. return False
  56. except Exception as e:
  57. logging.exception(f"An unexpected error occurred: {e}")
  58. return False
  61. if __name__ == '__main__':
  62. # Example Usage
  63. user_id = "user123"
  64. experiment_config_path = "experiment_config.json"
  65. script_path = "user_record_script.py"
  67. # Create dummy files for testing
  68. with open(experiment_config_path, "w") as f:
  69. json.dump({"input_parameters": {"param1": "value1"}}, f)
  70. with open(script_path, "w") as f:
  71. f.write("""
  72. import sys
  73. user_id = sys.argv[1]
  74. print(f"Processing user: {user_id}")
  75. print(f"Parameter 1: {sys.argv[2]}")
  76. """)
  78. success = bootstrap_user_record(user_id, experiment_config_path, script_path)
  79. if success:
  80. print(f"User record script bootstrapped successfully for {user_id}")
  81. else:
  82. print(f"Failed to bootstrap user record script for {user_id}")

Add your comment