Websocket server implementing a clock handling alarms and timezones
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. import shlex
  2. from argparse import Namespace
  3. from typing import Callable
  4. from .clock import Clock
  5. from .errors import DuplicatedCommandError, ArgumentParserError
  6. from .parser import CommandParser
  7. class Command(object):
  8. _commands = dict()
  9. def __init__(self, name:str, run_fun:Callable[[Clock, Namespace], str],
  10. **argument_parser_kwargs:dict):
  11. """ Register a new command
  12. @param name : The command
  13. @param run_fun : The callable handling the command (taking two
  14. arguments, the Clock instance and parsed args)
  15. @throws DuplicatedCommandError on duplicated name
  16. """
  17. if name in self.__class__._commands:
  18. msg = 'A command named %r allready exists' % name
  19. raise DuplicatedCommandError(msg)
  20. self.__name = name
  21. self.__parser = CommandParser(prog=name, **argument_parser_kwargs)
  22. self.__run = run_fun
  23. self.__class__._commands[self.__name] = self
  24. @property
  25. def name(self) -> str:
  26. """ The command name """
  27. return self.__name
  28. @property
  29. def description(self) -> str:
  30. """ The command description """
  31. return self.__parser.description
  32. @property
  33. def usage(self) -> str:
  34. """ The command usage """
  35. return self.__parser.format_usage().strip()
  36. @property
  37. def help(self) -> str:
  38. """ The command help text """
  39. return self.__parser.format_help()
  40. @staticmethod
  41. def fmt_ok(message:str='') -> str:
  42. """ Format a success command result message """
  43. return 'OK:%s' % message
  44. @staticmethod
  45. def fmt_err(reason:str) -> str:
  46. """ Format an error command result message """
  47. return 'ERR:%s' % reason
  48. @classmethod
  49. def list(cls) -> list[str]:
  50. """ Return the list of all command names """
  51. return list(cls._commands.keys())
  52. @classmethod
  53. def all(cls) -> dict:
  54. """ Return a dict associating command name and Command instance """
  55. return {k:v for k,v in cls._commands.items()}
  56. @classmethod
  57. def run_command(cls, clock:Clock, command:str) -> str:
  58. """ Run a command on the given Clock instance
  59. @param clock : The Clock instance
  60. @param command : A string with the command name and its arguments
  61. """
  62. try:
  63. argv = shlex.split(command)
  64. except ValueError as expt:
  65. return cls.fmt_err(expt)
  66. if len(argv) == 0:
  67. return cls.fmt_err('Empty command')
  68. if argv[0] not in cls._commands:
  69. return cls.fmt_err('Unknown command %r' % argv[0])
  70. return cls._commands[argv[0]]._run(clock, argv)
  71. def add_argument(self, name:str, *args, **kwargs):
  72. """ Add an argument to the CommandParser instance.
  73. Same signature than argparse.ArgumentParser.add_argument()
  74. """
  75. return self.__parser.add_argument(name, *args, **kwargs)
  76. def add_subparsers(self, *args,**kwargs):
  77. """ Add a subparser to the command's ArgumentParser """
  78. return self.__parser.add_subparsers(*args, **kwargs)
  79. def _run(self, clock:Clock, argv:list) -> str:
  80. """ Run the command on the given Clock instance
  81. @param clock : The Clock instance
  82. @param message : The command and its arguments
  83. @return Command result as text (formatted by one of the
  84. Command.fmt_ok() or Command.fmt_err() commands)
  85. """
  86. try:
  87. args = self.__parser.parse_args(argv[1:])
  88. except ArgumentParserError as expt:
  89. return self.fmt_err(expt)
  90. return self.__run(clock, args)