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

1"""ha-repl - run Python against a live Home Assistant. 

2 

3 ha-repl interactive API client shell (default) 

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

5 ha-repl custom interactive custom-component shell 

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

7 needs the ha_repl_server component 

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

9 ha-repl exec -f snippet.py run a file 

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

11 ha-repl reset / ha-repl sessions manage custom-mode server-side sessions 

12 

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

14Both HASS_SERVER/HASS_TOKEN also fall back to a `.env` file in the current directory, below real env vars. 

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

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

17""" 

18 

19from __future__ import annotations 

20 

21import argparse 

22import asyncio 

23import json 

24import os 

25import shutil 

26import sys 

27from typing import Any 

28 

29from .client import Client, HaReplError, resolve_token, resolve_url 

30 

31 

32def main() -> None: 

33 args = _parser().parse_args() 

34 try: 

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

36 except HaReplError as err: 

37 print(f"ha-repl: {err}", file=sys.stderr) 

38 sys.exit(2) 

39 except KeyboardInterrupt: 

40 sys.exit(130) 

41 

42 

43def _parser() -> argparse.ArgumentParser: 

44 parser = argparse.ArgumentParser( 

45 prog="ha-repl", 

46 description=__doc__, 

47 formatter_class=argparse.RawDescriptionHelpFormatter, 

48 ) 

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

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

51 parser.add_argument( 

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

53 ) 

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

55 parser.add_argument( 

56 "--ttl", 

57 type=float, 

58 default=30.0, 

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

60 ) 

61 parser.add_argument( 

62 "--no-auto-await", 

63 action="store_true", 

64 help="don't automatically await a call you forgot to `await` - " 

65 "report it as an unawaited coroutine instead, as plain Python would", 

66 ) 

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

68 

69 exec_ = sub.add_parser( 

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

71 ) 

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

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

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

75 exec_.add_argument( 

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

77 ) 

78 

79 sub.add_parser( 

80 "api", 

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

82 ) 

83 sub.add_parser( 

84 "custom", 

85 help="interactive custom-component shell: hass + obj, needs the ha_repl_server component", 

86 ) 

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

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

89 return parser 

90 

91 

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

93 url = resolve_url(args.server) 

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

95 match args.command: 

96 case "exec": 

97 return await _exec(client, args) 

98 case "reset": 

99 result = await client.call("ha_repl_server/reset", session=args.session) 

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

101 return 0 

102 case "sessions": 

103 result = await client.call("ha_repl_server/sessions") 

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

105 return 0 

106 case "custom": 

107 from .repl import run_repl 

108 

109 return await run_repl( 

110 client, args.session, auto_await=not args.no_auto_await 

111 ) 

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

113 from .apirepl import run_api_repl 

114 

115 return await run_api_repl( 

116 client, args.ttl, auto_await=not args.no_auto_await 

117 ) 

118 

119 

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

121 if args.file: 

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

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

124 code = fp.read() 

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

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

127 raise HaReplError("exec needs code, -f FILE, or - to read stdin") 

128 code = sys.stdin.read() 

129 else: 

130 code = args.code 

131 if args.reset: 

132 await client.call("ha_repl_server/reset", session=args.session) 

133 payload: dict[str, Any] = { 

134 "code": code, 

135 "session": args.session, 

136 "auto_await": not args.no_auto_await, 

137 **display_options(), 

138 } 

139 if args.timeout: 

140 payload["timeout"] = args.timeout 

141 if args.json: 

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

143 # that asked for machine-readable output. 

144 payload["color"] = False 

145 result = await client.call("ha_repl_server/exec", **payload) 

146 if args.json: 

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

148 else: 

149 print_result(result) 

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

151 

152 

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

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

155 

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

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

158 person watching it live. 

159 """ 

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

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

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

163 

164 

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

166 if result["stdout"]: 

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

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

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

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

171 print(result["value"]) 

172 if result["error"]: 

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

174 if result.get("truncated"): 

175 print("[ha-repl: output truncated]", file=sys.stderr) 

176 sys.stdout.flush() 

177 

178 

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

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

181 

182 

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

184 if not sessions: 

185 return "no sessions" 

186 return "\n".join( 

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

188 for s in sessions 

189 )