1#!/usr/bin/env python3

2 import re

3 import sys

4 import time

5 import logging

6 import argparse

7 from pathlib import Path

8 from datetime import datetime

9

10

11 class GPIOController:

12

13 GPIO_ROOT_PATH = "/sys/class/gpio"

14 GPIO_EXPORT_PATH = "{}/export".format(GPIO_ROOT_PATH)

15 GPIO_UNEXPORT_PATH = "{}/unexport".format(GPIO_ROOT_PATH)

16

17 def __init__(

18 self, gpiochip: str, gpiopin: str, direction: int, need_export: bool

19 ):

20 if gpiochip.isnumeric() is False or gpiopin.isnumeric() is False:

21 raise ValueError("Invalid GPIO chip or GPIO pin")

22

23 self._gpio_root_node = Path(self.GPIO_ROOT_PATH)

24 self._gpiochip_mapping = self.get_gpiochip_mapping()

25

26 if gpiochip not in self._gpiochip_mapping.keys():

27 raise KeyError("GPIO chip number {} is incorrect".format(gpiochip))

28

29 self.gpio_chip_node = self._gpio_root_node.joinpath(

30 "gpiochip{}".format(self._gpiochip_mapping.get(gpiochip))

31 )

32 self.gpio_node = self.value_node = None

33 self.gpiochip_info = {"base": None, "ngpio": None, "offset": gpiopin}

34 self._direction = direction

35 self._need_export = need_export

36 self.initial_state = {"value": None, "direction": None, "number": None}

37

38 def __enter__(self):

39 self.setup()

40 return self

41

42 def __exit__(self, type, value, traceback):

43 self.teardown()

44

45 def check_gpio_offset(self, pin: str, ngpio: str):

46 if int(pin) == 0:

47 raise ValueError("")

48

49 if int(pin) > int(ngpio):

50 raise IndexError(

51 "GPIO pin '{}' greater than ngpio value '{}'".format(

52 pin, ngpio

53 )

54 )

55

56 def get_gpiochip_mapping(self):

57 mapping = {}

58 nodes = sorted(self._gpio_root_node.glob("gpiochip*/device/gpiochip*"))

59 for node in nodes:

60 match = re.search(

61 r"/sys/class/gpio/gpiochip([0-9]+)/device/gpiochip([0-9]+)",

62 str(node),

63 )

64 if match: 64 ↛ 59 line 64 didn't jump to line 59 because the condition on line 64 was always true

65 mapping.update({match.groups()[1]: match.groups()[0]})

66 return mapping

67

68 def setup(self):

69 logging.debug("setup action for GPIO testing")

70 for key in ["base", "ngpio"]:

71 with self.gpio_chip_node.joinpath(key) as gpio_node:

72 if self._node_exists(gpio_node) is False:

73 raise FileNotFoundError(

74 "{} file not exists".format(str(gpio_node))

75 )

76 self.gpiochip_info[key] = self._read_node(gpio_node)

77

78 self.check_gpio_offset(

79 self.gpiochip_info["offset"], self.gpiochip_info["ngpio"]

80 )

81

82 self.initial_state["number"] = str(

83 (

84 int(self.gpiochip_info["base"])

85 + int(self.gpiochip_info["offset"])

86 - 1

87 )

88 )

89 self.gpio_node = self._gpio_root_node.joinpath(

90 "gpio{}".format(self.initial_state["number"])

91 )

92

93 # Export GPIO node if needed

94 if self._need_export:

95 self._export(self.initial_state["number"])

96 time.sleep(1)

97

98 if self._node_exists(self.gpio_node) is False:

99 raise FileNotFoundError(

100 "{} file not exists".format(str(self.gpio_node))

101 )

102

103 # Store the initial state for GPIO

104 self.initial_state["value"] = self.value

105 self.initial_state["direction"] = self.direction

106

107 # Configure the GPIO direction

108 self.direction = self._direction

109

110 def teardown(self):

111 logging.debug("teardown action for LED testing")

112 self.value = self.initial_state["value"]

113 self.direction = self.initial_state["direction"]

114 if self._need_export:

115 self._unexport(self.initial_state["number"])

116

117 def _node_exists(self, node: Path):

118 if node.exists() is False:

119 raise FileNotFoundError("{} file not exists".format(str(node)))

120

121 def _read_node(self, node: Path):

122 self._node_exists(node)

123 return node.read_text().strip("\n")

124

125 def _write_node(self, node: Path, value: str, check=True):

126

127 self._node_exists(node)

128 node.write_text(value)

129 if check and self._read_node(node) != value:

130 raise ValueError(

131 "Unable to change the value of {} file".format(str(node))

132 )

133

134 def _export(self, gpio_number: str):

135 logging.debug("export GPIO node %s", gpio_number)

136 with Path(self.GPIO_EXPORT_PATH) as gpio_node:

137 self._write_node(gpio_node, gpio_number, False)

138

139 def _unexport(self, gpio_number: str):

140 logging.debug("unexport GPIO node %s", gpio_number)

141 with Path(self.GPIO_UNEXPORT_PATH) as gpio_node:

142 self._write_node(gpio_node, gpio_number, False)

143

144 @property

145 def direction(self):

146 with self.gpio_node.joinpath("direction") as gpio_node:

147 return self._read_node(gpio_node)

148

149 @direction.setter

150 def direction(self, value: str):

151 if value not in ["in", "out"]:

152 raise ValueError(

153 "The {} is not allowed for direction".format(value)

154 )

155

156 with self.gpio_node.joinpath("direction") as gpio_node:

157 logging.debug(

158 "set direction to {} for {}".format(value, gpio_node.name)

159 )

160 self._write_node(gpio_node, value)

161

162 @property

163 def value(self):

164 with self.gpio_node.joinpath("value") as gpio_node:

165 return self._read_node(gpio_node)

166

167 @value.setter

168 def value(self, value: str):

169 if value not in ["1", "0"]:

170 raise ValueError("The {} is not allowed for value".format(value))

171

172 with self.gpio_node.joinpath("value") as gpio_node:

173 logging.debug(

174 "set value to {} for {}".format(value, gpio_node.name)

175 )

176 self._write_node(gpio_node, value)

177

178 def on(self):

179 logging.debug("turn on GPIO")

180 self.value = "1"

181

182 def off(self):

183 logging.debug("turn off GPIO")

184 self.value = "0"

185

186 def blinking(self, duration=10, interval=1):

187 logging.debug(

188 "set GPIO{} LED blinking".format(self.initial_state["number"])

189 )

190 start_time = datetime.now()

191 while (datetime.now() - start_time).total_seconds() <= duration:

192 self.on()

193 time.sleep(interval)

194 self.off()

195 time.sleep(interval)

196

197

198 def blinking_test(args):

199

200 with GPIOController(

201 args.gpio_chip, args.gpio_pin, "out", args.need_export

202 ) as led_controller:

203 logging.info(

204 (

205 "# Set the {} LED blinking around {} seconds "

206 "with {} seconds blink interval"

207 ).format(args.name, args.duration, args.interval)

208 )

209 led_controller.blinking(args.duration, args.interval)

210

211

212 def dump_gpiochip(args):

213 gpio_debug_path = "/sys/kernel/debug/gpio"

214

215 gpio_debug = Path(gpio_debug_path)

216 if gpio_debug.exists():

217 print(gpio_debug.read_text())

218 else:

219 raise FileNotFoundError("{} file not exists".format(str(gpio_debug)))

220

221

222 def leds_resource(args):

223 output = ""

224 resource_text = "name: {}\nchip_number: {}\nport: {}\n\n"

225 for led in args.mapping.split():

226 name, chip_number, port = led.split(":")

227 output += resource_text.format(name, chip_number, port)

228 print(output)

229

230

231 def register_arguments():

232 parser = argparse.ArgumentParser(

233 formatter_class=argparse.RawDescriptionHelpFormatter,

234 description="GPIO Control Tests",

235 )

236 parser.add_argument(

237 "--debug",

238 action="store_true",

239 help="Turn on debug level output for extra info during test run.",

240 )

241

242 sub_parsers = parser.add_subparsers(

243 help="GPIO test type", dest="test_func"

244 )

245 sub_parsers.required = True

246

247 gpio_led_parser = sub_parsers.add_parser("led")

248 gpio_led_parser.add_argument("-n", "--name", required=True, type=str)

249 gpio_led_parser.add_argument("-d", "--duration", type=int, default=5)

250 gpio_led_parser.add_argument("-i", "--interval", type=int, default=0.5)

251 gpio_led_parser.add_argument("--gpio-chip", type=str, required=True)

252 gpio_led_parser.add_argument("--gpio-pin", type=str, required=True)

253 gpio_led_parser.add_argument(

254 "--need-export", action="store_true", default=False

255 )

256 gpio_led_parser.set_defaults(test_func=blinking_test)

257

258 gpio_dump_parser = sub_parsers.add_parser("dump")

259 gpio_dump_parser.set_defaults(test_func=dump_gpiochip)

260

261 gpio_args_parser = sub_parsers.add_parser("led-resource")

262 gpio_args_parser.set_defaults(test_func=leds_resource)

263 gpio_args_parser.add_argument(

264 "mapping",

265 help=(

266 "Usage of parameter: GPIO_CONTROLLER_LEDS="

267 "{name1}:{controller1}:{port1} "

268 "{name2}:{controller1}:{port2} ..."

269 ),

270 )

271

272 args = parser.parse_args()

273 return args

274

275

276 if __name__ == "__main__":

277

278 root_logger = logging.getLogger()

279 root_logger.setLevel(logging.INFO)

280 logger_format = "%(asctime)s %(levelname)-8s %(message)s"

281 date_format = "%Y-%m-%d %H:%M:%S"

282

283 # Log DEBUG and INFO to stdout, others to stderr

284 stdout_handler = logging.StreamHandler(sys.stdout)

285 stdout_handler.setFormatter(logging.Formatter(logger_format, date_format))

286

287 stderr_handler = logging.StreamHandler(sys.stderr)

288 stderr_handler.setFormatter(logging.Formatter(logger_format, date_format))

289

290 stdout_handler.setLevel(logging.DEBUG)

291 stderr_handler.setLevel(logging.WARNING)

292

293 # Add a filter to the stdout handler to limit log records to

294 # INFO level and below

295 stdout_handler.addFilter(lambda record: record.levelno <= logging.INFO)

296

297 root_logger.addHandler(stderr_handler)

298 root_logger.addHandler(stdout_handler)

299

300 args = register_arguments()

301

302 if args.debug:

303 root_logger.setLevel(logging.DEBUG)

304

305 try:

306 args.test_func(args)

307 except Exception as err:

308 logging.error(err)