1#!/usr/bin/env python3

2"""

3 script to test accerometer functionality

4

5 Copyright (C) 2012 Canonical Ltd.

6

7 Authors

8 Jeff Marcom <jeff.marcom@canonical.com>

9

10 This program is free software: you can redistribute it and/or modify

11 it under the terms of the GNU General Public License version 3,

12 as published by the Free Software Foundation.

13

14 This program is distributed in the hope that it will be useful,

15 but WITHOUT ANY WARRANTY; without even the implied warranty of

16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17 GNU General Public License for more details.

18

19 You should have received a copy of the GNU General Public License

20 along with this program. If not, see <http://www.gnu.org/licenses/>.

21

22 The purpose of this script is to simply interact with an onboard

23 accelerometer, and check to be sure that the x, y, z axis respond

24 to physical movement of hardware.

25"""

26

27 from argparse import ArgumentParser

28 import gi

29 import logging

30 import os

31 import re

32 import sys

33 import threading

34 import time

35

36 gi.require_version("Gdk", "3.0")

37 gi.require_version("GLib", "2.0")

38 gi.require_version("Gtk", "3.0")

39 from gi.repository import Gdk, GLib, Gtk # noqa: E402

40 from subprocess import Popen, PIPE, check_output, STDOUT # noqa: E402

41 from subprocess import CalledProcessError # noqa: E402

42 from checkbox_support.parsers.modinfo import ModinfoParser # noqa: E402

43

44 handler = logging.StreamHandler()

45 logger = logging.getLogger()

46 logger.addHandler(handler)

47 logger.setLevel(logging.DEBUG)

48

49

50 class AccelerometerUI(Gtk.Window):

51 """Builds UI Framework for axis threshold tests using Gtk"""

52

53 def __init__(self):

54 Gtk.Window.__init__(self, title="Accelerometer Test")

55

56 self.set_default_size(450, 100)

57 self.set_type_hint(Gdk.WindowType.TOPLEVEL)

58

59 self.enabled = False

60

61 # Create UI Grid

62 w_table = Gtk.Grid()

63 self.add(w_table)

64

65 # Create axis buttons

66 self.up_icon = Gtk.Image(stock=Gtk.STOCK_GO_UP)

67 self.up_icon.set_padding(10, 30)

68 self.down_icon = Gtk.Image(stock=Gtk.STOCK_GO_DOWN)

69 self.down_icon.set_padding(10, 30)

70 self.left_icon = Gtk.Image(stock=Gtk.STOCK_GO_BACK)

71 self.right_icon = Gtk.Image(stock=Gtk.STOCK_GO_FORWARD)

72

73 # Set debug

74 self.debug_label = Gtk.Label("Debug")

75

76 # Set Grid layout for UI

77 message = "Please tilt your hardware in the positions shown below:"

78 w_table.attach(Gtk.Label(message), 0, 0, 4, 1)

79

80 w_table.attach(self.up_icon, 2, 2, 1, 1)

81 w_table.attach_next_to(

82 self.debug_label, self.up_icon, Gtk.PositionType.BOTTOM, 1, 1

83 )

84 w_table.attach_next_to(

85 self.down_icon, self.debug_label, Gtk.PositionType.BOTTOM, 1, 1

86 )

87 w_table.attach_next_to(

88 self.left_icon, self.debug_label, Gtk.PositionType.LEFT, 1, 1

89 )

90 w_table.attach_next_to(

91 self.right_icon, self.debug_label, Gtk.PositionType.RIGHT, 1, 1

92 )

93

94 def update_axis_icon(self, direction):

95 """Change desired directional icon to checkmark"""

96 exec(

97 "self.%s_icon.set_from_stock" % (direction)

98 + "(Gtk.STOCK_YES, size=Gtk.IconSize.BUTTON)"

99 )

100

101 def update_debug_label(self, text):

102 """Update axis information in center of UI"""

103 self.debug_label.set_text(text)

104

105 def destroy(self):

106 Gtk.main_quit()

107

108 def enable(self):

109 self.enabled = True

110 self.connect("delete-event", Gtk.main_quit)

111 self.show_all()

112

113 # Enable GLib/Gdk threading so the UI won't lock main

114 GLib.threads_init()

115 Gdk.threads_init()

116 Gdk.threads_enter()

117 Gtk.main()

118 Gdk.threads_leave()

119

120

121 class PermissionException(RuntimeError):

122 def __init__(self, error):

123 message = "Please re-run with root permissions: %s" % error.strip()

124 super(PermissionException, self).__init__(message)

125

126

127 class AxisData(threading.Thread):

128 """Acquire information from kernel regarding the state of the

129 accelerometer axis positions. Gathered data will be compared to

130 a preset threshold reading. The default threshold (either - or + )

131 for any direction is 600. Return values for thread are SUCCESS:0

132 FAILURE:1. FAILURE is likely to exists when thread is unable to

133 obtain a valid reading from the hardware."""

134

135 def __init__(self, device_path, ui_control=None):

136 threading.Thread.__init__(self)

137 self.ui = ui_control

138 self.device_path = device_path.strip("/")

139 self.tilt_threshold = 600

140 self.x_test_pool = ["up", "down"]

141 self.y_test_pool = ["left", "right"]

142

143 if self.ui is None:

144 self.ui.enabled = False

145

146 def grab_current_readings(self):

147 """Search device path and return axis tuple"""

148 time.sleep(0.5) # Sleep to accomodate slower processors

149 data_file = os.path.join(

150 "/sys", self.device_path, "device", "position"

151 )

152

153 # Try and retrieve positional data from kernel

154 try:

155 position_tuple = open(data_file)

156 except (OSError, IOError):

157 logging.error("Failed to open: %s" % data_file)

158 return False

159

160 # Split data for x, y, z as it's easier to manage threshold tests.

161 axis_set = position_tuple.read().strip("\n()")

162

163 return axis_set.split(",")

164

165 def parse_reading(self, value, mapping):

166 """Check for positive or negative threshold match"""

167 if abs(value) >= abs(self.tilt_threshold):

168 # And return test pool array position based on integer

169 if value < 0:

170 return 2

171 return 1

172

173 def direction_poll(self, x_axis, y_axis):

174 """Poll for threshold being met per x, and y axis"""

175 direction_map = {"X": x_axis, "Y": y_axis}

176

177 for mapping, data in direction_map.items():

178 reading = self.parse_reading(int(data), mapping)

179

180 if isinstance(reading, int):

181 return reading, mapping

182

183 # Return nothing if threshold is not met

184 return False, None

185

186 def run(self):

187 rem_tests = self.y_test_pool + self.x_test_pool

188 while len(rem_tests) > 0:

189 axis_data_bundle = self.grab_current_readings()

190

191 if not isinstance(axis_data_bundle, list):

192 logging.error("Failed to grab appropriate readings")

193 return 1

194

195 # Parse for current positional values

196 # Hdaps will only report X, and Y positional data

197 x_data = int(axis_data_bundle[0])

198 y_data = int(axis_data_bundle[1])

199 if len(axis_data_bundle) > 2:

200 z_data = int(axis_data_bundle[2])

201 else:

202 z_data = 0

203

204 debug_info = "X: %s Y: %s Z: %s" % (x_data, y_data, z_data)

205

206 if self.ui.enabled:

207 # Update positional values in UI

208 self.ui.update_debug_label(debug_info)

209

210 position, axis = self.direction_poll(x_data, y_data)

211 if position:

212 # Check axis set and delete test from pool

213 if axis == "X":

214 pool = self.x_test_pool

215 else:

216 pool = self.y_test_pool

217 if len(pool) >= position:

218 direction = pool[position - 1]

219 if direction in rem_tests:

220 # Remove direction from test pool

221 del rem_tests[rem_tests.index(direction)]

222 self.ui.update_axis_icon(direction)

223 else:

224 # Accept readings as successful test result

225 logging.debug("Latest Readings: %s" % debug_info)

226 break

227

228 if self.ui.enabled:

229 self.ui.destroy()

230 return 0

231

232

233 def insert_supported_module(oem_module):

234 """Try and insert supported module to see if we get any init errors"""

235 try:

236 stream = check_output(

237 ["modinfo", oem_module], stderr=STDOUT, universal_newlines=True

238 )

239 except CalledProcessError as err:

240 print("Error accessing modinfo for %s: " % oem_module, file=sys.stderr)

241 print(err.output, file=sys.stderr)

242 return err.returncode

243

244 parser = ModinfoParser(stream)

245 module = os.path.basename(parser.get_field("filename"))

246

247 insmod_output = Popen(

248 ["insmod %s" % module],

249 stderr=PIPE,

250 shell=True,

251 universal_newlines=True,

252 )

253

254 error = insmod_output.stderr.read()

255 if "Permission denied" in error:

256 raise PermissionException(error)

257

258 return insmod_output.returncode

259

260

261 def check_module_status():

262 """Looks to see if it can determine the hardware manufacturer

263 and report corresponding accelerometer driver status"""

264 oem_driver_pool = {

265 "hewlett-packard": "hp_accel",

266 "toshiba": "hp_accel",

267 "ibm": "hdaps",

268 "lenovo": "hdaps",

269 }

270

271 oem_module = None

272 dmi_info = Popen(

273 ["dmidecode"], stdout=PIPE, stderr=PIPE, universal_newlines=True

274 )

275

276 output, error = dmi_info.communicate()

277

278 if "Permission denied" in error:

279 raise PermissionException(error)

280

281 vendor_data = re.findall(r"Vendor:\s.*", output)

282 try:

283 manufacturer = vendor_data[0].split(":")[1].strip()

284 except IndexError:

285 logging.error("Failed to find Manufacturing data")

286 return

287

288 logging.debug(manufacturer)

289

290 # Now we look to see if there was any info during boot

291 # time that would help in debugging this failure

292 for vendor, module in oem_driver_pool.items():

293 if manufacturer.lower() == vendor:

294 oem_module = oem_driver_pool.get(vendor)

295 break # We've found our desired module to probe.

296

297 if oem_module is not None:

298 if insert_supported_module(oem_module) is not None:

299 logging.error("Failed module insertion")

300 # Check dmesg status for supported module

301 driver_status = Popen(["dmesg"], stdout=PIPE, universal_newlines=True)

302 module_regex = oem_module + ".*"

303 kernel_notes = re.findall(module_regex, driver_status.stdout.read())

304 # Report ALL findings, it's useful to note it the driver failed init

305 # more than once of actually passed despite a reading failure

306 logging.debug("\n".join((kernel_notes)))

307 else:

308 logging.error("No supported module")

309

310

311 def check_for_accelerometer():

312 """Checks device list for existence of accelerometer and returns

313 name, manufacturer, and system path info."""

314

315 found = False

316 device_info = open("/proc/bus/input/devices").readlines()

317 for line in device_info:

318 if "accelerometer" in line.lower():

319 target = device_info.index(line)

320

321 name = device_info[target].split("=")[1]

322 path = device_info[target + 2].split("=")[1]

323 found = True

324 break

325

326 if found:

327 logger.debug("Name: %s\nPath: %s" % (name, path))

328 return path.strip()

329 else:

330 # Return False as it's expected

331 logger.error("Accelerometer hardware not found")

332

333 return False

334

335

336 def main():

337

338 parser = ArgumentParser(description="Tests accelerometer functionality")

339

340 parser.add_argument(

341 "-m",

342 "--manual",

343 default=False,

344 action="store_true",

345 help="For manual test with visual notification",

346 )

347 parser.add_argument(

348 "-a",

349 "--automated",

350 default=True,

351 action="store_true",

352 help="For automated test using defined parameters",

353 )

354

355 args = parser.parse_args()

356

357 sys_path = check_for_accelerometer()

358 if not sys_path:

359 try:

360 check_module_status()

361 except PermissionException as error:

362 print(error, file=sys.stderr)

363 sys.exit(1)

364

365 ui = AccelerometerUI()

366 grab_data = AxisData(sys_path, ui)

367 grab_data.setDaemon(True)

368 grab_data.start()

369

370 if args.manual:

371 ui.enable()

372 else:

373 # Sleep for enough time to retrieve a reading.

374 # Reading is not instant.

375 time.sleep(5)

376

377

378 if __name__ == "__main__":

379 main()