interactive_prompt.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. from framework.manager import Manager
  2. from state.devstate import DevState
  3. from InquirerPy import prompt, separator
  4. from pprint import pprint
  5. import prompt_questions as questions
  6. from inspect import signature
  7. from uuid import UUID
  8. from ast import literal_eval
  9. def generate_context_question(ctx_type, services):
  10. """
  11. Converts service names to human readable form
  12. """
  13. choices = [
  14. s.__name__.replace('_', ' ') for s in services
  15. ]
  16. choices = sorted(choices)
  17. choices.append(separator.Separator())
  18. choices.append("close context")
  19. ctx_question = [
  20. {
  21. 'type': 'list',
  22. 'name': 'op',
  23. 'message': f'Currently in context {ctx_type.__name__}, which operation would you like to perform?',
  24. 'choices': choices,
  25. 'filter': lambda x: x.replace(' ', '_')
  26. }
  27. ]
  28. return ctx_question
  29. def main():
  30. state = DevState()
  31. man = Manager(state)
  32. while True:
  33. if man.current_model != None and man.current_context == None:
  34. # we have selected a model, so we display typing questions
  35. answer = prompt(questions.MODEL_SELECTED)
  36. ctx = man
  37. elif man.current_model != None and man.current_context != None:
  38. # we have selected both a model and a context, so we display available services
  39. qs = generate_context_question(type(man.current_context), man.get_services())
  40. answer = prompt(qs)
  41. if answer['op'] == 'close_context':
  42. man.close_context()
  43. continue
  44. else:
  45. ctx = man.current_context
  46. else:
  47. answer = prompt(questions.MODEL_MGMT)
  48. ctx = man
  49. if answer['op'] == 'exit':
  50. break
  51. else:
  52. method = getattr(ctx, answer['op'])
  53. args_questions = []
  54. types = {}
  55. for p in signature(method).parameters.values():
  56. types[p.name] = p.annotation if p.annotation else literal_eval # can't use filter in question dict, doesn't work for some reason...
  57. if p.annotation == UUID:
  58. args_questions.append({
  59. 'type': 'list',
  60. 'name': p.name,
  61. 'message': f'{p.name.replace("_", " ")}?',
  62. 'choices': list(man.get_models()),
  63. 'filter': lambda x: state.read_value(state.read_dict(state.read_root(), x))
  64. })
  65. else:
  66. args_questions.append({
  67. 'type': 'input',
  68. 'name': p.name,
  69. 'message': f'{p.name.replace("_", " ")}?',
  70. 'filter': lambda x: '' if x.lower() == 'false' else x
  71. })
  72. args = prompt(args_questions)
  73. args = {k: types[k](v) if len(v) > 0 else None for k, v in args.items()}
  74. try:
  75. output = method(**args)
  76. if output != None:
  77. try:
  78. if isinstance(output, str):
  79. raise TypeError
  80. output = list(output)
  81. if len(output) > 0:
  82. for o in sorted(output):
  83. print(f"\u2022 {o}")
  84. except TypeError:
  85. print(f"\u2022 {output}")
  86. except RuntimeError as e:
  87. print(e)
  88. if __name__ == '__main__':
  89. print("""Welcome to...\r\n __ ____ _____ \r\n | \\/ \\ \\ / /__ \\ \r\n | \\ / |\\ \\ / / ) |\r\n | |\\/| | \\ \\/ / / / \r\n | | | | \\ / / /_ \r\n |_| |_| \\/ |____| """)
  90. main()