Coverage for _build/src/powerprofilesctl: 82%

218 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2025-03-09 21:27 +0000

1#!/usr/bin/python3 

2 

3import argparse 

4import os 

5import signal 

6import subprocess 

7import sys 

8from gi.repository import Gio, GLib 

9 

10PP_NAME = "org.freedesktop.UPower.PowerProfiles" 

11PP_PATH = "/org/freedesktop/UPower/PowerProfiles" 

12PP_IFACE = "org.freedesktop.UPower.PowerProfiles" 

13PROPERTIES_IFACE = "org.freedesktop.DBus.Properties" 

14 

15 

16def get_proxy(): 

17 bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) 

18 return Gio.DBusProxy.new_sync( 

19 bus, Gio.DBusProxyFlags.NONE, None, PP_NAME, PP_PATH, PROPERTIES_IFACE, None 

20 ) 

21 

22 

23def command(func): 

24 def wrapper(*args, **kwargs): 

25 try: 

26 func(*args, **kwargs) 

27 except GLib.Error as error: 

28 sys.stderr.write( 

29 f"Failed to communicate with power-profiles-daemon: {error}\n" 

30 ) 

31 sys.exit(1) 

32 except ValueError as error: 

33 sys.stderr.write(f"Error: {error}\n") 

34 sys.exit(1) 

35 

36 return wrapper 

37 

38 

39@command 

40def _version(_args): 

41 client_version = "0.30" 

42 try: 

43 proxy = get_proxy() 

44 daemon_ver = proxy.Get("(ss)", PP_IFACE, "Version") 

45 except GLib.Error: 

46 daemon_ver = "unknown" 

47 print(f"client: {client_version}\ndaemon: {daemon_ver}") 

48 

49 

50@command 

51def _set_profile(args): 

52 proxy = get_proxy() 

53 proxy.Set( 

54 "(ssv)", PP_IFACE, "ActiveProfile", GLib.Variant.new_string(args.profile[0]) 

55 ) 

56 

57 

58@command 

59def _get(_args): 

60 proxy = get_proxy() 

61 profile = proxy.Get("(ss)", PP_IFACE, "ActiveProfile") 

62 print(profile) 

63 

64 

65@command 

66def _set_battery_aware(args): 

67 enable = args.enable 

68 disable = args.disable 

69 if enable is False and disable is True: 

70 raise ValueError("enable or disable is required") 

71 if enable is True and disable is False: 

72 raise ValueError("can't set both enable and disable") 

73 enable = enable if enable is not None else not disable 

74 proxy = get_proxy() 

75 proxy.Set("(ssv)", PP_IFACE, "BatteryAware", GLib.Variant.new_boolean(enable)) 

76 

77 

78def get_profiles_property(prop): 

79 proxy = get_proxy() 

80 return proxy.Get("(ss)", PP_IFACE, prop) 

81 

82 

83def get_profile_choices(): 

84 try: 

85 return [profile["Profile"] for profile in get_profiles_property("Profiles")] 

86 except GLib.Error: 

87 return [] 

88 

89 

90@command 

91def _list(_args): 

92 profiles = get_profiles_property("Profiles") 

93 reason = get_proxy().Get("(ss)", PP_IFACE, "PerformanceDegraded") 

94 degraded = reason != "" 

95 active = get_proxy().Get("(ss)", PP_IFACE, "ActiveProfile") 

96 

97 index = 0 

98 for profile in reversed(profiles): 

99 if index > 0: 

100 print("") 

101 marker = "*" if profile["Profile"] == active else " " 

102 print(f'{marker} {profile["Profile"]}:') 

103 for driver in ["CpuDriver", "PlatformDriver"]: 

104 if driver not in profile: 

105 continue 

106 value = profile[driver] 

107 print(f" {driver}:\t{value}") 

108 if profile["Profile"] == "performance": 

109 print(" Degraded: ", f"yes ({reason})" if degraded else "no") 

110 index += 1 

111 

112 

113@command 

114def _list_holds(_args): 

115 holds = get_profiles_property("ActiveProfileHolds") 

116 

117 index = 0 

118 for hold in holds: 

119 if index > 0: 

120 print("") 

121 print("Hold:") 

122 print(" Profile: ", hold["Profile"]) 

123 print(" Application ID: ", hold["ApplicationId"]) 

124 print(" Reason: ", hold["Reason"]) 

125 index += 1 

126 

127 

128@command 

129def _launch(args): 

130 reason = args.reason 

131 profile = args.profile 

132 appid = args.appid 

133 if not args.arguments: 

134 raise ValueError("No command to launch") 

135 if not args.appid: 

136 appid = args.arguments[0] 

137 if not profile: 

138 profile = "performance" 

139 if not reason: 

140 reason = f"Running {args.appid}" 

141 ret = 0 

142 bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) 

143 proxy = Gio.DBusProxy.new_sync( 

144 bus, Gio.DBusProxyFlags.NONE, None, PP_NAME, PP_PATH, PP_IFACE, None 

145 ) 

146 cookie = proxy.HoldProfile("(sss)", profile, reason, appid) 

147 

148 # print (f'Got {cookie} for {profile} hold') 

149 with subprocess.Popen(args.arguments) as launched_app: 

150 # Redirect the same signal to the child 

151 def receive_signal(signum, _stack): 

152 launched_app.send_signal(signum) 

153 

154 redirected_signals = [ 

155 signal.SIGTERM, 

156 signal.SIGINT, 

157 signal.SIGABRT, 

158 ] 

159 

160 for sig in redirected_signals: 

161 signal.signal(sig, receive_signal) 

162 

163 try: 

164 launched_app.wait() 

165 ret = launched_app.returncode 

166 except KeyboardInterrupt: 

167 ret = launched_app.returncode 

168 

169 for sig in redirected_signals: 

170 signal.signal(sig, signal.SIG_DFL) 

171 

172 proxy.ReleaseProfile("(u)", cookie) 

173 

174 if ret < 0: 

175 # Use standard POSIX signal exit code. 

176 os.kill(os.getpid(), -ret) 

177 return 

178 

179 sys.exit(ret) 

180 

181 

182@command 

183def _query_battery_aware(_args): 

184 result = get_profiles_property("BatteryAware") 

185 print(f"Dynamic changes from charger and battery events: {result}") 

186 

187 

188@command 

189def _list_actions(_args): 

190 actions = get_profiles_property("ActionsInfo") 

191 for action in actions: 

192 for key in action: 

193 print(f"{key}: {action[key]}") 

194 if action != actions[-1]: 

195 print("") 

196 

197 

198@command 

199def _configure_action(args): 

200 action = args.action[0] 

201 enable = args.enable 

202 disable = args.disable 

203 if enable is False and disable is True: 

204 raise argparse.ArgumentError( 

205 argument="action", message="enable or disable is required" 

206 ) 

207 if enable is True and disable is False: 

208 raise argparse.ArgumentError( 

209 argument="action", message="can't set both enable and disable" 

210 ) 

211 print(f"action: {action}, enable: {enable}") 

212 bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) 

213 proxy = Gio.DBusProxy.new_sync( 

214 bus, Gio.DBusProxyFlags.NONE, None, PP_NAME, PP_PATH, PP_IFACE, None 

215 ) 

216 proxy.SetActionEnabled("(sb)", action, enable) 

217 

218 

219def get_parser(): 

220 parser = argparse.ArgumentParser( 

221 epilog="Use “powerprofilesctl COMMAND --help” to get detailed help for individual commands", 

222 ) 

223 subparsers = parser.add_subparsers(help="Individual command help", dest="command") 

224 parser_list = subparsers.add_parser("list", help="List available power profiles") 

225 parser_list.set_defaults(func=_list) 

226 parser_list_holds = subparsers.add_parser( 

227 "list-holds", help="List current power profile holds" 

228 ) 

229 parser_list_holds.set_defaults(func=_list_holds) 

230 parser_list_actions = subparsers.add_parser( 

231 "list-actions", help="List available power profile actions" 

232 ) 

233 parser_list_actions.set_defaults(func=_list_actions) 

234 parser_get = subparsers.add_parser( 

235 "get", help="Print the currently active power profile" 

236 ) 

237 parser_get.set_defaults(func=_get) 

238 parser_set = subparsers.add_parser( 

239 "set", help="Set the currently active power profile" 

240 ) 

241 parser_set.add_argument( 

242 "profile", 

243 nargs=1, 

244 help="Profile to use for set command", 

245 choices=get_profile_choices(), 

246 ) 

247 parser_set.set_defaults(func=_set_profile) 

248 parser_set_action = subparsers.add_parser( 

249 "configure-action", help="Configure the action to be taken for the profile" 

250 ) 

251 parser_set_action.add_argument( 

252 "action", 

253 nargs=1, 

254 help="action to change for configure-action", 

255 ) 

256 parser_set_action.add_argument( 

257 "--enable", 

258 action="store_true", 

259 help="enable action", 

260 ) 

261 parser_set_action.add_argument( 

262 "--disable", 

263 action="store_false", 

264 help="disable action", 

265 ) 

266 parser_set_action.set_defaults(func=_configure_action) 

267 parser_set_battery_aware = subparsers.add_parser( 

268 "configure-battery-aware", 

269 help="Turn on or off dynamic changes from battery level or power adapter", 

270 ) 

271 parser_set_battery_aware.add_argument( 

272 "--enable", 

273 action="store_true", 

274 help="enable battery aware", 

275 ) 

276 parser_set_battery_aware.add_argument( 

277 "--disable", 

278 action="store_false", 

279 help="disable battery aware", 

280 ) 

281 parser_set_battery_aware.set_defaults(func=_set_battery_aware) 

282 parser_query_battery_aware = subparsers.add_parser( 

283 "query-battery-aware", 

284 help="Query if dynamic changes from battery level or power adapter are enabled", 

285 ) 

286 parser_query_battery_aware.set_defaults(func=_query_battery_aware) 

287 parser_launch = subparsers.add_parser( 

288 "launch", 

289 help="Launch a command while holding a power profile", 

290 description="Launch the command while holding a power profile, " 

291 "either performance, or power-saver. By default, the profile hold " 

292 "is for the performance profile, but it might not be available on " 

293 "all systems. See the list command for a list of available profiles.", 

294 ) 

295 parser_launch.add_argument( 

296 "arguments", 

297 nargs="*", 

298 help="Command to launch", 

299 ) 

300 parser_launch.add_argument( 

301 "--profile", "-p", required=False, help="Profile to use for launch command" 

302 ) 

303 parser_launch.add_argument( 

304 "--reason", "-r", required=False, help="Reason to use for launch command" 

305 ) 

306 parser_launch.add_argument( 

307 "--appid", "-i", required=False, help="AppId to use for launch command" 

308 ) 

309 parser_launch.set_defaults(func=_launch) 

310 parser_version = subparsers.add_parser( 

311 "version", help="Print version information and exit" 

312 ) 

313 parser_version.set_defaults(func=_version) 

314 

315 if not os.getenv("PPD_COMPLETIONS_GENERATION"): 

316 return parser 

317 

318 try: 

319 import shtab # pylint: disable=import-outside-toplevel 

320 

321 shtab.add_argument_to(parser, ["--print-completion"]) # magic! 

322 except ImportError: 

323 pass 

324 

325 return parser 

326 

327 

328def check_unknown_args(args, unknown_args, cmd): 

329 if cmd != "launch": 

330 return False 

331 

332 for idx, unknown_arg in enumerate(unknown_args): 

333 arg = args[idx] 

334 if arg == cmd: 

335 return True 

336 if unknown_arg == arg: 

337 return False 

338 

339 return True 

340 

341 

342def main(): 

343 parser = get_parser() 

344 args, unknown = parser.parse_known_args() 

345 # default behavior is to run list if no command is given 

346 if not args.command: 

347 args.func = _list 

348 

349 if check_unknown_args(sys.argv[1:], unknown, args.command): 

350 args.arguments += unknown 

351 unknown = [] 

352 

353 if unknown: 

354 msg = argparse._("unrecognized arguments: %s") 

355 parser.error(msg % " ".join(unknown)) 

356 

357 args.func(args) 

358 

359 

360if __name__ == "__main__": 

361 main()