1#!/usr/bin/env python3

2#

3# This file is part of Checkbox.

4#

5# Copyright 2008-2024 Canonical Ltd.

6# Written by:

7# Matt Fischer <matt@mattfischer.com>

8# Sylvain Pineau <sylvain.pineau@canonical.com>

9# Fernando Bravo <fernando.bravo.hernandez@canonical.com>

10#

11# The v4l2 ioctl code comes from the Python bindings for the v4l2

12# userspace api (http://pypi.python.org/pypi/v4l2):

13# Copyright (C) 1999-2009 the contributors

14#

15# The JPEG metadata parser is a part of bfg-pages:

16# http://code.google.com/p/bfg-pages/source/browse/trunk/pages/getimageinfo.py

17# Copyright (C) Tim Hoffman

18#

19# Checkbox is free software: you can redistribute it and/or modify

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

21# as published by the Free Software Foundation.

22#

23# Checkbox is distributed in the hope that it will be useful,

24# but WITHOUT ANY WARRANTY; without even the implied warranty of

25# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

26# GNU General Public License for more details.

27#

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

29# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

30#

31

32 import argparse

33 import ctypes

34 import errno

35 import fcntl

36 import logging

37 import os

38 import re

39 import struct

40 import sys

41

42 from glob import glob

43 from subprocess import check_call, CalledProcessError, STDOUT

44 from tempfile import NamedTemporaryFile

45

46

47_IOC_NRBITS = 8

48_IOC_TYPEBITS = 8

49_IOC_SIZEBITS = 14

50

51_IOC_NRSHIFT = 0

52_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS

53_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS

54_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS

55

56_IOC_WRITE = 1

57_IOC_READ = 2

58

59

60 def _IOC(dir_, type_, nr, size):

61 return (

62 ctypes.c_int32(dir_ << _IOC_DIRSHIFT).value

63 | ctypes.c_int32(ord(type_) << _IOC_TYPESHIFT).value

64 | ctypes.c_int32(nr << _IOC_NRSHIFT).value

65 | ctypes.c_int32(size << _IOC_SIZESHIFT).value

66 )

67

68

69 def _IOC_TYPECHECK(t):

70 return ctypes.sizeof(t)

71

72

73 def _IOR(type_, nr, size):

74 return _IOC(_IOC_READ, type_, nr, ctypes.sizeof(size))

75

76

77 def _IOWR(type_, nr, size):

78 return _IOC(_IOC_READ | _IOC_WRITE, type_, nr, _IOC_TYPECHECK(size))

79

80

81 class v4l2_capability(ctypes.Structure):

82 """

83 Driver capabilities

84 """

85

86 _fields_ = [

87 ("driver", ctypes.c_char * 16),

88 ("card", ctypes.c_char * 32),

89 ("bus_info", ctypes.c_char * 32),

90 ("version", ctypes.c_uint32),

91 ("capabilities", ctypes.c_uint32),

92 ("reserved", ctypes.c_uint32 * 4),

93 ]

94

95

96# Values for 'capabilities' field

97 V4L2_CAP_VIDEO_CAPTURE = 0x00000001

98 V4L2_CAP_VIDEO_CAPTURE_MPLANE = 0x00001000

99 V4L2_CAP_VIDEO_OVERLAY = 0x00000004

100 V4L2_CAP_READWRITE = 0x01000000

101 V4L2_CAP_STREAMING = 0x04000000

102

103 v4l2_frmsizetypes = ctypes.c_uint

104(

105 V4L2_FRMSIZE_TYPE_DISCRETE,

106 V4L2_FRMSIZE_TYPE_CONTINUOUS,

107 V4L2_FRMSIZE_TYPE_STEPWISE,

108) = range(1, 4)

109

110

111 class v4l2_frmsize_discrete(ctypes.Structure):

112 _fields_ = [

113 ("width", ctypes.c_uint32),

114 ("height", ctypes.c_uint32),

115 ]

116

117

118 class v4l2_frmsize_stepwise(ctypes.Structure):

119 _fields_ = [

120 ("min_width", ctypes.c_uint32),

121 ("max_width", ctypes.c_uint32),

122 ("step_width", ctypes.c_uint32),

123 ("min_height", ctypes.c_uint32),

124 ("max_height", ctypes.c_uint32),

125 ("step_height", ctypes.c_uint32),

126 ]

127

128

129 class v4l2_frmsizeenum(ctypes.Structure):

130 class _u(ctypes.Union):

131 _fields_ = [

132 ("discrete", v4l2_frmsize_discrete),

133 ("stepwise", v4l2_frmsize_stepwise),

134 ]

135

136 _fields_ = [

137 ("index", ctypes.c_uint32),

138 ("pixel_format", ctypes.c_uint32),

139 ("type", ctypes.c_uint32),

140 ("_u", _u),

141 ("reserved", ctypes.c_uint32 * 2),

142 ]

143

144 _anonymous_ = ("_u",)

145

146

147 class v4l2_fmtdesc(ctypes.Structure):

148 _fields_ = [

149 ("index", ctypes.c_uint32),

150 ("type", ctypes.c_int),

151 ("flags", ctypes.c_uint32),

152 ("description", ctypes.c_char * 32),

153 ("pixelformat", ctypes.c_uint32),

154 ("reserved", ctypes.c_uint32 * 4),

155 ]

156

157

158 V4L2_FMT_FLAG_COMPRESSED = 0x0001

159 V4L2_FMT_FLAG_EMULATED = 0x0002

160

161

162# ioctl code for video devices

163 VIDIOC_QUERYCAP = _IOR("V", 0, v4l2_capability)

164 VIDIOC_ENUM_FRAMESIZES = _IOWR("V", 74, v4l2_frmsizeenum)

165 VIDIOC_ENUM_FMT = _IOWR("V", 2, v4l2_fmtdesc)

166

167

168 class CameraTest:

169 """

170 This class is handles all the different camera tests. The tests available

171 are:

172 - detect: Display information regarding webcam hardware

173 - led: Activate camera (switch on led), but don't display any output

174 - video: Displays the preview window for a video stream

175 - image: Captures an image to a file

176 - resolutions: After querying the webcam for supported formats and

177 resolutions, take multiple images using the first format returned

178 by the driver, and see if they are valid.

179 """

180

181 def __init__(self, **kwargs):

182 self._width = 640

183 self._height = 480

184 self._devices = []

185 self.device = kwargs.get("device", "/dev/video0")

186 self.headless = kwargs.get("headless", False)

187 self.output = kwargs.get("output", "")

188 self.log_level = kwargs.get("log_level", logging.INFO)

189

190 self.main_loop = None

191 self.pipeline = None

192 self.timeout = None

193

194 def init_gstreamer(self):

195 """

196 Initialize the GStreamer and GLib libraries

197 """

198 import gi

199

200 gi.require_version("Gst", "1.0")

201 from gi.repository import Gst

202

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

204 from gi.repository import GLib

205

206 self.Gst = Gst

207 self.GLib = GLib

208

209 self.Gst.init(None)

210

211 def init_gtk(self):

212 """

213 Initialize the Gtk library

214 """

215 import gi

216

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

218 from gi.repository import Gtk

219

220 self.Gtk = Gtk

221 self.Gtk.init([])

222

223 def detect(self):

224 """

225 Display information regarding webcam hardware

226 """

227 cap_status = dev_status = 1

228 for i in range(10):

229 cp = v4l2_capability()

230 device = "/dev/video%d" % i

231 try:

232 with open(device, "r") as vd:

233 fcntl.ioctl(vd, VIDIOC_QUERYCAP, cp)

234 except IOError:

235 continue

236 dev_status = 0

237 cap_status = self._detect_and_show_camera_info(device, cp)

238

239 return dev_status | cap_status

240

241 def _detect_and_show_camera_info(self, device, cp):

242 """

243 Detect the capture capability and show the information of a specific

244 camera device

245

246 :param device:

247 Full path of camera device under /dev. e.g. /dev/video0

248 :param cp:

249 The v4l2 capability

250

251 :returns:

252 0 if the camera supports the capture capability

253 1 if the camera doesn't support the capture capability

254 """

255 capture_capabilities = (

256 cp.capabilities & V4L2_CAP_VIDEO_CAPTURE

257 or cp.capabilities & V4L2_CAP_VIDEO_CAPTURE_MPLANE

258 )

259

260 print("%s: OK" % device)

261 print(" name : %s" % cp.card.decode("UTF-8"))

262 print(" driver : %s" % cp.driver.decode("UTF-8"))

263 print(

264 " version: %s.%s.%s"

265 % (cp.version >> 16, (cp.version >> 8) & 0xFF, cp.version & 0xFF)

266 )

267 print(

268 " flags : 0x%x [" % cp.capabilities,

269 " CAPTURE" if capture_capabilities else "",

270 " OVERLAY" if cp.capabilities & V4L2_CAP_VIDEO_OVERLAY else "",

271 " READWRITE" if cp.capabilities & V4L2_CAP_READWRITE else "",

272 " STREAMING" if cp.capabilities & V4L2_CAP_STREAMING else "",

273 " ]",

274 sep="",

275 )

276

277 formats = self._supported_formats_to_string(

278 self._get_supported_formats(device)

279 )

280 formats = formats.replace("Resolutions:", " Resolutions:")

281 formats = formats.replace("Format:", " Format:")

282 print(formats)

283

284 return 0 if capture_capabilities else 1

285

286 def _on_gst_message(self, bus, message):

287 # Process End-of-stream messages

288 if message.type == self.Gst.MessageType.EOS:

289 # Stop the pipeline

290 self.pipeline.set_state(self.Gst.State.NULL)

291 # Quit the GLib main loop

292 self.main_loop.quit()

293

294 logging.debug("End-of-stream")

295

296 # Process Error messages

297 elif message.type == self.Gst.MessageType.ERROR:

298 # Stop the pipeline

299 self.pipeline.set_state(self.Gst.State.NULL)

300 # Quit the GLib main loop

301 self.main_loop.quit()

302 err, debug = message.parse_error()

303 print("Error: {}".format(err.message))

304

305 logging.debug("Debug info: {}".format(debug))

306

307 raise SystemExit("Error: {}".format(err.message))

308

309 # Process Pipeline state change messages

310 elif ( 310 ↛ exit line 310 didn't return from function '_on_gst_message' because the condition on line 310 was always true

311 message.type == self.Gst.MessageType.STATE_CHANGED

312 and message.src.get_name() == "pipeline"

313 ):

314 old_state, new_state, _ = message.parse_state_changed()

315 logging.debug(

316 "Pipeline changed state from {} to {}".format(

317 old_state.value_nick, new_state.value_nick

318 )

319 )

320

321 def _stop_pipeline(self):

322 """

323 Stop the GLib main loop and set the pipeline state to NULL

324 """

325 self.main_loop.quit()

326 self.pipeline.set_state(self.Gst.State.NULL)

327

328 def _on_timeout(self):

329 """

330 Stop the pipeline when the timeout is reached and remove the timeout

331 """

332 print("Timeout exceeded")

333 self.timeout = None

334 self._stop_pipeline()

335

336 def led(self):

337 """

338 Activate camera (switch on led), but don't display any output

339 """

340 self.main_loop = self.GLib.MainLoop()

341 self.GLib.timeout_add_seconds(3, self._stop_pipeline)

342 self._setup_video_gstreamer("fakesink")

343

344 def video(self):

345 """

346 Displays the preview window for a video stream

347 """

348 # Don't display the video, just run the camera

349 if self.headless:

350 self.main_loop = self.GLib.MainLoop()

351 self.GLib.timeout_add_seconds(4, self._stop_pipeline)

352 self._setup_video_gstreamer("fakesink")

353 else:

354 print("Starting video preview")

355 self.GLib.timeout_add_seconds(10, self._stop_pipeline)

356 self._setup_video_gstreamer()

357

358 def _setup_video_gstreamer(self, sink=None):

359 """

360 Setup the gstreamer pipeline to create the video stream

361 """

362 webcam = self.Gst.ElementFactory.make("v4l2src")

363 webcam.set_property("device", self.device)

364 wrappercamerabinsrc = self.Gst.ElementFactory.make(

365 "wrappercamerabinsrc"

366 )

367 wrappercamerabinsrc.set_property("video-source", webcam)

368 pipeline = self.Gst.ElementFactory.make("camerabin", "pipeline")

369 pipeline.set_property("camera-source", wrappercamerabinsrc)

370 if sink:

371 vf_sink = self.Gst.ElementFactory.make(sink)

372 pipeline.set_property("viewfinder-sink", vf_sink)

373 pipeline.set_state(self.Gst.State.PAUSED)

374 caps = pipeline.get_property("viewfinder-supported-caps")

375 supported_resolutions = {}

376 for i in range(caps.get_size()):

377 key = caps.get_structure(i).get_int("width").value

378 if key not in supported_resolutions.keys(): 378 ↛ 380 line 378 didn't jump to line 380 because the condition on line 378 was always true

379 supported_resolutions[key] = set()

380 supported_resolutions[key].add(

381 caps.get_structure(i).get_int("height").value

382 )

383 if not supported_resolutions:

384 raise SystemExit("No supported resolutions found!")

385 width = min(

386 supported_resolutions.keys(), key=lambda x: abs(x - self._width)

387 )

388 height = min(

389 supported_resolutions[width], key=lambda y: abs(y - self._height)

390 )

391 vf_caps = self.Gst.Caps.from_string(

392 "video/x-raw, width={}, height={}".format(width, height)

393 )

394 pipeline.set_property("viewfinder-caps", vf_caps)

395

396 bus = pipeline.get_bus()

397 bus.add_signal_watch()

398

399 self.pipeline = pipeline

400 bus.connect("message", self._on_gst_message)

401 self.pipeline.set_state(self.Gst.State.PLAYING)

402

403 self.main_loop = self.GLib.MainLoop()

404 try:

405 self.main_loop.run()

406 except self.GLib.Error:

407 self.main_loop.quit()

408 self.pipeline.set_state(self.Gst.State.NULL)

409

410 def image(self):

411 """

412 Captures an image to a file

413 """

414 pixelformat = self._get_default_format()["pixelformat"]

415 if self.output:

416 self._capture_image(

417 self.output, self._width, self._height, pixelformat

418 )

419 else:

420 with NamedTemporaryFile(

421 prefix="camera_test_", suffix=".jpg", delete=False

422 ) as f:

423 self._capture_image(

424 f.name, self._width, self._height, pixelformat

425 )

426

427 def _capture_image(self, filename, width, height, pixelformat):

428 """

429 Captures an image to a given filename. If the image capture fails with

430 fswebcam, it will try to capture the image with gstreamer.

431 """

432 # Try to take a picture with fswebcam

433 if not self._capture_image_fswebcam(

434 filename, width, height, pixelformat

435 ):

436 print("Failed to capture image with fswebcam, using gstreamer")

437 # If fswebcam fails, try with gstreamer

438 self._capture_image_gstreamer(filename, width, height, pixelformat)

439 print("Image saved to %s" % filename)

440 if not self.headless:

441 self._display_image(filename, width, height)

442

443 def _capture_image_fswebcam(self, filename, width, height, pixelformat):

444 """

445 Simple wrapper around fswebcam to capture an image

446 """

447 command = [

448 "fswebcam",

449 "-D 1",

450 "-S 50",

451 "--no-banner",

452 "-d",

453 self.device,

454 "-r",

455 "%dx%d" % (width, height),

456 filename,

457 ]

458 if pixelformat: 458 ↛ 463 line 458 didn't jump to line 463 because the condition on line 458 was always true

459 # special tweak for fswebcam

460 command.extend(

461 ["-p", pixelformat if pixelformat != "MJPG" else "MJPEG"]

462 )

463 try:

464 check_call(command, stdout=open(os.devnull, "w"), stderr=STDOUT)

465 return os.path.getsize(filename) != 0

466 except (CalledProcessError, OSError):

467 return False

468

469 def _capture_image_gstreamer(self, filename, width, height, pixelformat):

470 """

471 Setup the gstreamer pipeline to capture an image.

472 This pipeline consists of the following elements:

473 - v4l2src: Capture video from a V4L2 device

474 - caps: Set the video stream to the desired resolution

475 - rgb_capture: last element in the pipeline to capture the image. If

476 no conversion is needed, this is the same as caps

477 - jpegenc: Encode the video stream to a JPEG image

478 - filesink: Save the JPEG image to a file

479 """

480

481 pipeline = self.Gst.Pipeline.new("pipeline")

482

483 # Add source

484 source = self.Gst.ElementFactory.make("v4l2src", "video-source")

485 source.set_property("device", self.device)

486 source.set_property("num-buffers", 10)

487 pipeline.add(source)

488

489 # Add the caps element and include a format converter if needed to

490 # always get a RGB format.

491 caps = self.Gst.ElementFactory.make("capsfilter", "caps")

492 if pixelformat == "RG10":

493 caps.set_property(

494 "caps",

495 self.Gst.Caps.from_string(

496 "video/x-bayer, format=gbrg, width={},height={}".format(

497 width, height

498 )

499 ),

500 )

501 # Add bayer2rgb

502 bayer2rgb = self.Gst.ElementFactory.make("bayer2rgb", "bayer2rgb")

503

504 pipeline.add(caps)

505 pipeline.add(bayer2rgb)

506 caps.link(bayer2rgb)

507 rgb_capture = bayer2rgb

508

509 else:

510 caps.set_property(

511 "caps",

512 self.Gst.Caps.from_string(

513 "video/x-raw,width={},height={}".format(width, height)

514 ),

515 )

516 pipeline.add(caps)

517 rgb_capture = caps

518

519 # Add encoder

520 encoder = self.Gst.ElementFactory.make("jpegenc", "encoder")

521 pipeline.add(encoder)

522

523 # Add sink

524 sink = self.Gst.ElementFactory.make("filesink", "sink")

525 sink.set_property("location", filename)

526 pipeline.add(sink)

527

528 # Link elements

529 source.link(caps)

530 rgb_capture.link(encoder)

531 encoder.link(sink)

532

533 # Connect the bus to the message handler

534 bus = pipeline.get_bus()

535 bus.add_signal_watch()

536 bus.connect("message", self._on_gst_message)

537

538 # Start the pipeline

539 self.pipeline = pipeline

540 self.pipeline.set_state(self.Gst.State.PLAYING)

541

542 # Add a timeout of 90 seconds to capture the image

543 self.timeout = self.GLib.timeout_add_seconds(90, self._on_timeout)

544

545 # Start the main loop. If the loop finishes successfully, we will

546 # remove the timeout. If the timeout is reached, we will stop the

547 # pipeline calling the _on_timeout method and stop both the

548 self.main_loop = self.GLib.MainLoop()

549 try:

550 self.main_loop.run()

551 except self.GLib.Error:

552 self.main_loop.quit()

553 self.pipeline.set_state(self.Gst.State.NULL)

554

555 # If the image is captured correctly and the timeout is not reached,

556 # remove the GLib timeout, so it does not interfere with the next

557 # iteration.

558 if self.timeout: 558 ↛ exit line 558 didn't return from function '_capture_image_gstreamer' because the condition on line 558 was always true

559 self.GLib.source_remove(self.timeout)

560

561 def _display_image(self, filename, width, height):

562 """

563 Display an image using Gtk

564 """

565 print("starting GTK")

566 # Initialize GTK application

567 window = self.Gtk.Window(title="Image Viewer")

568 window.set_default_size(width, height)

569

570 # Load and display the image

571 image = self.Gtk.Image.new_from_file(filename)

572 window.add(image)

573

574 # Connect the destroy event to quit the GTK main loop

575 window.connect("destroy", self.Gtk.main_quit)

576 self.GLib.timeout_add_seconds(10, self.Gtk.main_quit)

577

578 # Show all widgets in the window

579 window.show_all()

580

581 # Start the GTK main loop

582 self.Gtk.main()

583

584 def _supported_formats_to_string(self, supported_formats):

585 """

586 Return a printable string representing a list of supported resolutions

587 """

588 ret = ""

589 for format in supported_formats:

590 ret += "Format: %s (%s)\n" % (

591 format["pixelformat"],

592 format["description"],

593 )

594 ret += "Resolutions: "

595 for resolution in format["resolutions"]:

596 ret += "%sx%s," % (resolution[0], resolution[1])

597 # truncate the extra comma with :-1

598 ret = ret[:-1] + "\n"

599 return ret

600

601 def resolutions(self):

602 """

603 After querying the webcam for supported formats and resolutions,

604 take multiple images using the first format returned by the driver,

605 and see if they are valid

606 """

607 self.headless = True

608

609 format = self._get_default_format()

610 print(

611 "Taking multiple images using the %s pixelformat"

612 % format["pixelformat"]

613 )

614

615 if self.output: 615 ↛ 618 line 615 didn't jump to line 618 because the condition on line 615 was always true

616 self._save_debug_image(format, self.device, self.output)

617

618 for resolution in reversed(format["resolutions"]):

619 w = resolution[0]

620 h = resolution[1]

621 f = NamedTemporaryFile(

622 prefix="camera_test_%s%sx%s" % (format["pixelformat"], w, h),

623 suffix=".jpg",

624 delete=False,

625 )

626 print("Taking a picture at %sx%s" % (w, h))

627

628 self._capture_image(

629 f.name, w, h, pixelformat=format["pixelformat"]

630 )

631 if self._validate_image(f.name, w, h):

632 print("Validated image %s" % f.name)

633 os.remove(f.name)

634 else:

635 print("Failed to validate image %s" % f.name, file=sys.stderr)

636 os.remove(f.name)

637 return 1

638 return 0

639

640 def _save_debug_image(self, format, device, output):

641 """

642 Save an image to a file

643 """

644 # Check if the output directory exists

645 if not os.path.exists(output):

646 raise SystemExit(

647 "Output directory does not exist: {}".format(output)

648 )

649

650 # Choose one resolution image to store as an artifact. We will use

651 # the closest resolution to 640x480 as the target to have some

652 # uniformity in the output

653 target = 640

654 closest_resolution = min(

655 format["resolutions"], key=lambda x: abs(x[0] - target)

656 )

657 w, h = closest_resolution

658 device_name = device.split("/")[-1]

659 filepath = os.path.join(

660 output, "resolution_test_image_{}.jpg".format(device_name)

661 )

662 print("Saving debug image to %s" % filepath)

663 with open(filepath, "w") as f:

664 self._capture_image(f.name, w, h, format["pixelformat"])

665

666 def _get_supported_pixel_formats(self, device, maxformats=5):

667 """

668 Query the camera to see what pixel formats it supports. A list of

669 dicts is returned consisting of format and description. The caller

670 should check whether this camera supports VIDEO_CAPTURE before

671 calling this function.

672 """

673 supported_pixel_formats = []

674 fmt = v4l2_fmtdesc()

675 fmt.index = 0

676 fmt.type = V4L2_CAP_VIDEO_CAPTURE

677 try:

678 while fmt.index < maxformats:

679 with open(device, "r") as vd:

680 if fcntl.ioctl(vd, VIDIOC_ENUM_FMT, fmt) == 0:

681 pixelformat = {}

682 # save the int type for re-use later

683 pixelformat["pixelformat_int"] = fmt.pixelformat

684 pixelformat["pixelformat"] = "%s%s%s%s" % (

685 chr(fmt.pixelformat & 0xFF),

686 chr((fmt.pixelformat >> 8) & 0xFF),

687 chr((fmt.pixelformat >> 16) & 0xFF),

688 chr((fmt.pixelformat >> 24) & 0xFF),

689 )

690 pixelformat["description"] = fmt.description.decode()

691 supported_pixel_formats.append(pixelformat)

692 fmt.index = fmt.index + 1

693 except IOError as e:

694 # EINVAL is the ioctl's way of telling us that there are no

695 # more formats, so we ignore it

696 if e.errno != errno.EINVAL:

697 print(

698 "Unable to determine Pixel Formats, this may be a "

699 "driver issue."

700 )

701 return supported_pixel_formats

702 return supported_pixel_formats

703

704 def _get_supported_formats(self, device):

705 """

706 Query the camera for supported format info for a given pixel_format.

707 Data is returned in a list of dictionaries with supported pixel

708 formats as the following example shows:

709 format_info['pixelformat'] = "YUYV"

710 format_info['description'] = "(YUV 4:2:2 (YUYV))"

711 format_info['resolutions'] = [[width, height], [640, 480], [1280, 720]]

712

713 If we are unable to gather any information from the driver, then we

714 return YUYV and 640x480 which seems to be a safe default.

715 Per the v4l2 spec the ioctl used here is experimental

716 but seems to be well supported.

717 """

718 supported_formats_info = self._get_supported_pixel_formats(device)

719

720 # If we can't get any formats, we will return YUYV and 640x480

721 if not supported_formats_info:

722 format_info = {}

723 format_info["description"] = "YUYV"

724 format_info["pixelformat"] = "YUYV"

725 format_info["resolutions"] = [[640, 480]]

726 return [format_info]

727

728 for supported_format in supported_formats_info:

729 resolutions = []

730 framesize = v4l2_frmsizeenum()

731 framesize.index = 0

732 framesize.pixel_format = supported_format["pixelformat_int"]

733 with open(device, "r") as vd:

734 try:

735 while ( 735 ↛ 774 line 735 didn't jump to line 774

736 fcntl.ioctl(vd, VIDIOC_ENUM_FRAMESIZES, framesize) == 0

737 ):

738 if framesize.type == V4L2_FRMSIZE_TYPE_DISCRETE:

739 resolutions.append(

740 [

741 framesize.discrete.width,

742 framesize.discrete.height,

743 ]

744 )

745 # for continuous and stepwise, let's just use min and

746 # max they use the same structure and only return

747 # one result

748 elif framesize.type in ( 748 ↛ 765 line 748 didn't jump to line 765 because the condition on line 748 was always true

749 V4L2_FRMSIZE_TYPE_CONTINUOUS,

750 V4L2_FRMSIZE_TYPE_STEPWISE,

751 ):

752 resolutions.append(

753 [

754 framesize.stepwise.min_width,

755 framesize.stepwise.min_height,

756 ]

757 )

758 resolutions.append(

759 [

760 framesize.stepwise.max_width,

761 framesize.stepwise.max_height,

762 ]

763 )

764 break

765 framesize.index = framesize.index + 1

766 except IOError as e:

767 # EINVAL is the ioctl's way of telling us that there are no

768 # more formats, so we ignore it

769 if e.errno != errno.EINVAL:

770 print(

771 "Unable to determine supported framesizes "

772 "(resolutions), this may be a driver issue."

773 )

774 supported_format["resolutions"] = resolutions

775 return supported_formats_info

776

777 def _get_default_format(self):

778 formats = self._get_supported_formats(self.device)

779 # print supported formats and formats for the logs

780 print(self._supported_formats_to_string(formats))

781

782 # pick the first format, which seems to be what the driver wants for a

783 # default. This also matches the logic that fswebcam uses to select

784 # a default format.

785 if not formats:

786 raise SystemExit("No supported formats found")

787 format = formats[0]

788 return format

789

790 def _validate_image(self, filename, width, height):

791 """

792 Given a filename, ensure that the image is the width and height

793 specified and is a valid image file.

794 """

795 if not os.path.exists(filename):

796 print("Image file not found")

797 return False

798

799 outw = outh = 0

800 with open(filename, mode="rb") as f:

801

802 # Check if the header of the image to see if it's a valid JPEG file

803 header = f.read(32)

804 if ( 804 ↛ 812 line 804 didn't jump to line 812 because the condition on line 804 was always true

805 header[6:10] != b"JFIF"

806 and header[6:10] != b"Exif"

807 and header[6:11] != b"Adobe"

808 ):

809 print("Image is not a JPEG file")

810 return False

811

812 f.seek(2)

813 b = f.read(1)

814 try:

815 w = 0

816 h = 0

817 while b and ord(b) != 0xDA:

818 while ord(b) != 0xFF:

819 b = f.read(1)

820 while ord(b) == 0xFF:

821 b = f.read(1)

822 if ord(b) >= 0xC0 and ord(b) <= 0xC3:

823 f.seek(3, 1)

824 h, w = struct.unpack(">HH", f.read(4))

825 break

826 b = f.read(1)

827 outw, outh = int(w), int(h)

828 except (struct.error, ValueError):

829 pass

830

831 if outw == 0 or outh == 0:

832 print("Image dimensions not found in JPEG file")

833 return False

834 if outw != width:

835 print(

836 "Image width does not match, was %s should be %s"

837 % (outw, width),

838 file=sys.stderr,

839 )

840 return False

841 if outh != height:

842 print(

843 "Image width does not match, was %s should be %s"

844 % (outh, height),

845 file=sys.stderr,

846 )

847 return False

848

849 return True

850

851

852 def parse_arguments(argv):

853 """

854 Parse command line arguments

855 """

856 parser = argparse.ArgumentParser(description="Run a camera-related test")

857

858 # Add subparsers for the different tests

859 subparsers = parser.add_subparsers(

860 dest="test", title="test", description="Available camera tests"

861 )

862

863 # Add a debug option

864 parser.add_argument(

865 "--debug",

866 dest="log_level",

867 action="store_const",

868 const=logging.DEBUG,

869 default=logging.INFO,

870 help="Show debugging messages",

871 )

872

873 # Function to manage the device parameter, either by specifying the device

874 # or by using the highest or lowest available device

875 def add_device_parameter(parser):

876 group = parser.add_mutually_exclusive_group()

877 group.add_argument(

878 "-d",

879 "--device",

880 default="/dev/video0",

881 help="Device for the webcam to use",

882 )

883 group.add_argument(

884 "--highest-device",

885 action="store_true",

886 help=(

887 "Use the /dev/videoN where N is the highest value available"

888 ),

889 )

890 group.add_argument(

891 "--lowest-device",

892 action="store_true",

893 help=("Use the /dev/videoN where N is the lowest value available"),

894 )

895

896 # Detect subparser

897 subparsers.add_parser("detect")

898

899 # Led subparser

900 led_parser = subparsers.add_parser("led")

901 add_device_parameter(led_parser)

902

903 # Video subparser

904 video_parser = subparsers.add_parser("video")

905 add_device_parameter(video_parser)

906 # add a headless option, false by default

907 video_parser.add_argument(

908 "-hl",

909 "--headless",

910 action="store_true",

911 help=("Don't display video, just run the camera"),

912 )

913

914 # Image subparser

915 image_parser = subparsers.add_parser("image")

916 add_device_parameter(image_parser)

917 image_parser.add_argument(

918 "-o",

919 "--output",

920 default="",

921 help="Output directory to store the image",

922 )

923 image_parser.add_argument(

924 "-hl",

925 "--headless",

926 action="store_true",

927 help=("Don't display picture, just write the picture to a file"),

928 )

929

930 # Resolutions subparser

931 resolutions_parser = subparsers.add_parser("resolutions")

932 resolutions_parser.add_argument(

933 "-o",

934 "--output",

935 default="",

936 help="Output directory to store a small debug image",

937 )

938 add_device_parameter(resolutions_parser)

939

940 args = parser.parse_args(argv)

941

942 # Handle the selection of the highest or lowest device

943 def get_video_devices():

944 devices = sorted(

945 glob("/dev/video[0-9]"), key=lambda d: re.search(r"\d", d).group(0)

946 )

947 assert len(devices) > 0, "No video devices found"

948 return devices

949

950 if hasattr(args, "highest_device") and args.highest_device:

951 args.device = get_video_devices()[-1]

952 elif hasattr(args, "lowest_device") and args.lowest_device:

953 args.device = get_video_devices()[0]

954 return vars(args)

955

956

957 if __name__ == "__main__":

958 args = parse_arguments(sys.argv[1:])

959

960 # Set the default test to detect if not specified

961 if not args.get("test"):

962 args["test"] = "detect"

963

964 # Set the log level

965 logging.basicConfig(level=args["log_level"])

966

967 camera = CameraTest(**args)

968 # Import Gst only for the test cases that will need it

969 if args["test"] in ["video", "image", "led", "resolutions"]:

970 camera.init_gstreamer()

971

972 # Import Gtk only for the test cases that will need it

973 if args["test"] in ["video", "image"] and not args["headless"]:

974 camera.init_gtk()

975

976 sys.exit(getattr(camera, args["test"])())