Coverage for src/dev_shell/cli.py: 0%

1"""dev_shell - run Python against a live Home Assistant. 

2 

3 dev_shell interactive API client shell (default) 

4 `obj` only, read-only, no HACS component needed 

5 dev_shell custom interactive custom-component shell 

6 `hass` + `obj` (same `obj` API, live not cached); 

7 needs the dev_shell_server component 

8 dev_shell exec 'hass.states.get("sun.sun")' run a snippet (custom mode only) 

9 dev_shell exec -f snippet.py run a file 

10 dev_shell exec - <<'EOF' ... EOF read the snippet from stdin 

11 dev_shell reset / dev_shell sessions manage custom-mode server-side sessions 

12 

13Connection: HASS_SERVER (default http://homeassistant.local:8123), HASS_TOKEN, HASS_SESSION. 

14API client mode: --ttl seconds before the cached snapshot is refreshed (default 30). 

15Exit status of exec is 1 when the snippet raised, 2 on connection/usage errors. 

16""" 

17 

18from __future__ import annotations 

19 

20import argparse 

21import asyncio 

22import json 

23import os 

24import shutil 

25import sys 

26from typing import Any 

27 

28from .client import Client, DevShellError, resolve_token, resolve_url 

29 

30 

31def main() -> None: 

32 args = _parser().parse_args() 

33 try: 

34 sys.exit(asyncio.run(_dispatch(args))) 

35 except DevShellError as err: 

36 print(f"dev_shell: {err}", file=sys.stderr) 

37 sys.exit(2) 

38 except KeyboardInterrupt: 

39 sys.exit(130) 

40 

41 

42def _parser() -> argparse.ArgumentParser: 

43 parser = argparse.ArgumentParser( 

44 prog="dev_shell", 

45 description=__doc__, 

46 formatter_class=argparse.RawDescriptionHelpFormatter, 

47 ) 

48 parser.add_argument("--url", default=os.environ.get("HASS_SERVER")) 

49 parser.add_argument("--token", default=None, help="defaults to $HASS_TOKEN") 

50 parser.add_argument( 

51 "-s", "--session", default=os.environ.get("HASS_SESSION", "default") 

52 ) 

53 parser.add_argument("--json", action="store_true", help="print raw JSON results") 

54 parser.add_argument( 

55 "--ttl", 

56 type=float, 

57 default=30.0, 

58 help="API client mode: seconds before the cached snapshot is refreshed (default: 30)", 

59 ) 

60 sub = parser.add_subparsers(dest="command") 

61 

62 exec_ = sub.add_parser( 

63 "exec", help="run code and print output and result (custom mode only)" 

64 ) 

65 exec_.add_argument("code", nargs="?", help="code to run, or - for stdin") 

66 exec_.add_argument("-f", "--file", help="run the contents of a file") 

67 exec_.add_argument("-t", "--timeout", type=float, help="cancel after N seconds") 

68 exec_.add_argument( 

69 "--reset", action="store_true", help="reset the session before running" 

70 ) 

71 

72 sub.add_parser( 

73 "api", 

74 help="interactive API client shell (the default): obj only, no HACS component needed", 

75 ) 

76 sub.add_parser( 

77 "custom", 

78 help="interactive custom-component shell: hass + obj, needs the dev_shell_server component", 

79 ) 

80 sub.add_parser("reset", help="discard the custom-mode session's variables") 

81 sub.add_parser("sessions", help="list custom-mode sessions on the server") 

82 return parser 

83 

84 

85async def _dispatch(args: argparse.Namespace) -> int: 

86 url = resolve_url(args.url) 

87 async with Client(url, resolve_token(args.token)) as client: 

88 match args.command: 

89 case "exec": 

90 return await _exec(client, args) 

91 case "reset": 

92 result = await client.call( 

93 "dev_shell_server/reset", session=args.session 

94 ) 

95 _emit(args, result, "reset" if result["reset"] else "no such session") 

96 return 0 

97 case "sessions": 

98 result = await client.call("dev_shell_server/sessions") 

99 _emit(args, result, _format_sessions(result["sessions"])) 

100 return 0 

101 case "custom": 

102 from .repl import run_repl 

103 

104 return await run_repl(client, args.session) 

105 case _: # "api", or no subcommand at all - API client mode is the default 

106 from .apirepl import run_api_repl 

107 

108 return await run_api_repl(client, args.ttl) 

109 

110 

111async def _exec(client: Client, args: argparse.Namespace) -> int: 

112 if args.file: 

113 # One-shot CLI read, not a hot path - a threaded read would be overkill. 

114 with open(args.file, encoding="utf-8") as fp: # noqa: ASYNC230 

115 code = fp.read() 

116 elif args.code in (None, "-"): 

117 if args.code is None and sys.stdin.isatty(): 

118 raise DevShellError("exec needs code, -f FILE, or - to read stdin") 

119 code = sys.stdin.read() 

120 else: 

121 code = args.code 

122 if args.reset: 

123 await client.call("dev_shell_server/reset", session=args.session) 

124 payload: dict[str, Any] = { 

125 "code": code, 

126 "session": args.session, 

127 **display_options(), 

128 } 

129 if args.timeout: 

130 payload["timeout"] = args.timeout 

131 if args.json: 

132 # Escape codes embedded in a JSON string are just noise for a consumer 

133 # that asked for machine-readable output. 

134 payload["color"] = False 

135 result = await client.call("dev_shell_server/exec", **payload) 

136 if args.json: 

137 print(json.dumps(result, indent=2)) 

138 else: 

139 print_result(result) 

140 return 1 if result["error"] else 0 

141 

142 

143def display_options() -> dict[str, Any]: 

144 """Color/width hints for the server to render values and tracebacks with. 

145 

146 Decided here, not there: the server only sees a websocket, not a terminal, 

147 and stdout here might be piped (a script, a redirected log) rather than a 

148 person watching it live. 

149 """ 

150 color = sys.stdout.isatty() and not os.environ.get("NO_COLOR") 

151 width = shutil.get_terminal_size((88, 24)).columns 

152 return {"color": color, "width": width} 

153 

154 

155def print_result(result: dict[str, Any]) -> None: 

156 if result["stdout"]: 

157 sys.stdout.write(result["stdout"]) 

158 if not result["stdout"].endswith("\n"): 

159 sys.stdout.write("\n") 

160 if result["value"] is not None: 

161 print(result["value"]) 

162 if result["error"]: 

163 sys.stderr.write(result["error"]["traceback"]) 

164 if result.get("truncated"): 

165 print("[dev_shell: output truncated]", file=sys.stderr) 

166 sys.stdout.flush() 

167 

168 

169def _emit(args: argparse.Namespace, result: Any, text: str) -> None: 

170 print(json.dumps(result, indent=2) if args.json else text) 

171 

172 

173def _format_sessions(sessions: list[dict[str, Any]]) -> str: 

174 if not sessions: 

175 return "no sessions" 

176 return "\n".join( 

177 f"{s['name']}: {s['executions']} runs, vars: {', '.join(s['variables'])}" 

178 for s in sessions 

179 )