Coverage for src/homeassistant_repl/repl.py: 0%

1"""Interactive shell: local line editing and history, remote execution.""" 

2 

3from __future__ import annotations 

4 

5import ast 

6import codeop 

7from pathlib import Path 

8 

9from prompt_toolkit import PromptSession 

10from prompt_toolkit.history import FileHistory 

11from prompt_toolkit.lexers import PygmentsLexer 

12from pygments.lexers.python import PythonLexer 

13 

14from .cli import display_options, print_result 

15from .client import Client 

16 

17HISTORY = Path.home() / ".ha_repl_history" 

18 

19 

20async def run_repl( 

21 client: Client, session_name: str, *, auto_await: bool = True 

22) -> int: 

23 compiler = codeop.CommandCompiler() 

24 compiler.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

25 prompt: PromptSession[str] = PromptSession( 

26 history=FileHistory(str(HISTORY)), lexer=PygmentsLexer(PythonLexer) 

27 ) 

28 print( 

29 f"Home Assistant REPL connected to {client.url} (session {session_name!r}). Ctrl-D to exit." 

30 ) 

31 lines: list[str] = [] 

32 while True: 

33 try: 

34 line = await prompt.prompt_async("... " if lines else ">>> ") 

35 except KeyboardInterrupt: 

36 lines.clear() 

37 continue 

38 except EOFError: 

39 return 0 

40 lines.append(line) 

41 source = "\n".join(lines) 

42 try: 

43 # None means incomplete input: keep reading continuation lines. 

44 if compiler(source, "<ha_repl>", "single") is None: 

45 continue 

46 except SyntaxError, OverflowError, ValueError: 

47 pass # send it anyway so the server reports the error consistently 

48 lines.clear() 

49 if not source.strip(): 

50 continue 

51 if is_quit_call(source): 

52 return 0 

53 result = await client.call( 

54 "ha_repl_server/exec", 

55 code=source, 

56 session=session_name, 

57 auto_await=auto_await, 

58 **display_options(), 

59 ) 

60 print_result(result) 

61 

62 

63def is_quit_call(source: str) -> bool: 

64 """True for a bare `quit()` or `exit()` - the standard way to leave a REPL. 

65 

66 Sending that to the server would just raise SystemExit there, caught and 

67 reported back as a confusing traceback instead of actually exiting 

68 anything, so it's handled locally like Ctrl-D instead. 

69 """ 

70 try: 

71 (stmt,) = ast.parse(source).body 

72 except SyntaxError, ValueError: 

73 return False 

74 return ( 

75 isinstance(stmt, ast.Expr) 

76 and isinstance(stmt.value, ast.Call) 

77 and isinstance(stmt.value.func, ast.Name) 

78 and stmt.value.func.id in ("quit", "exit") 

79 and not stmt.value.args 

80 and not stmt.value.keywords 

81 )