Merge pull request #40 from tiagocoutinho/video-subdevice

Video subdevice + video event improvements

linuxpy/codegen/video.py

@@ -119,8 +119,10 @@ class StandardID(enum.IntFlag):
     CEnum("ControlWhichValue", "V4L2_CTRL_WHICH_"),
     CEnum("TimeCodeType", "V4L2_TC_TYPE_"),
     CEnum("TimeCodeFlag", "V4L2_TC_FLAG_", "IntFlag"),
-    CEnum("EventType", "V4L2_EVENT_"),
     CEnum("EventSubscriptionFlag", "V4L2_EVENT_SUB_FL_", "IntFlag"),
+    CEnum("EventControlChange", "V4L2_EVENT_CTRL_CH_"),
+    CEnum("EventType", "V4L2_EVENT_"),
+    CEnum("MbusFrameFormatFlag", "V4L2_MBUS_FRAMEFMT_", "IntFlag"),
     # It is very dificult to match just only these two values using prefix, so put whole name there
     CEnum("Interlaced", ["V4L2_DV_PROGRESSIVE", "V4L2_DV_INTERLACED"], with_prefix=True),
     # It is very dificult to match just only these two values using prefix, so put whole name there

linuxpy/video/device.py

@@ -61,6 +61,7 @@ def _enum(name, prefix, klass=enum.IntEnum):
 ControlClass = raw.ControlClass
 SelectionTarget = raw.SelectionTarget
 EventType = raw.EventType
+EventControlChange = raw.EventControlChange
 IOC = raw.IOC
 BufferType = raw.BufType
 BufferFlag = raw.BufferFlag
@@ -731,7 +732,7 @@ def set_priority(fd, priority: Priority):
 
 def subscribe_event(
     fd,
-    event_type: EventType = EventType.ALL,
+    event_type: EventType,
     id: int = 0,
     flags: EventSubscriptionFlag = 0,
 ):
@@ -749,10 +750,9 @@ def unsubscribe_event(fd, event_type: EventType = EventType.ALL, id: int = 0):
     ioctl(fd, IOC.UNSUBSCRIBE_EVENT, sub)
 
 
-def deque_event(fd):
+def deque_event(fd) -> raw.v4l2_event:
     event = raw.v4l2_event()
-    ioctl(fd, IOC.DQEVENT, event)
-    return event
+    return ioctl(fd, IOC.DQEVENT, event)
 
 
 def set_edid(fd, edid):
@@ -821,6 +821,32 @@ def query_std(fd) -> StandardID:
     return StandardID(out.value)
 
 
+SubdevFormat = collections.namedtuple(
+    "SubdevFormat", "pad which width height code field colorspace quantization xfer_func flags stream"
+)
+
+
+def _translate_subdev_format(fmt: raw.v4l2_subdev_format):
+    return SubdevFormat(
+        pad=fmt.pad,
+        which=raw.SubdevFormatWhence(fmt.which),
+        width=fmt.format.width,
+        height=fmt.format.height,
+        code=raw.MbusPixelcode(fmt.format.code),
+        field=raw.Field(fmt.format.field),
+        colorspace=raw.Colorspace(fmt.format.colorspace),
+        quantization=raw.Quantization(fmt.format.quantization),
+        xfer_func=raw.XferFunc(fmt.format.xfer_func),
+        flags=raw.MbusFrameFormatFlag(fmt.format.flags),
+        stream=fmt.stream,
+    )
+
+
+def get_subdevice_format(fd, pad: int = 0) -> raw.v4l2_subdev_format:
+    fmt = raw.v4l2_subdev_format(pad=pad, which=raw.SubdevFormatWhence.ACTIVE)
+    return _translate_subdev_format(ioctl(fd, IOC.SUBDEV_G_FMT, fmt))
+
+
 # Helpers
 
 
@@ -983,6 +1009,11 @@ def query_std(self) -> StandardID:
         return query_std(self.fileno())
 
 
+class SubDevice(BaseDevice):
+    def get_format(self, pad: int = 0) -> SubdevFormat:
+        return get_subdevice_format(self, pad=pad)
+
+
 def create_artificial_control_class(class_id):
     return raw.v4l2_query_ext_ctrl(
         id=class_id | 1,
@@ -1950,6 +1981,10 @@ async def __aiter__(self):
         while True:
             yield await self.aread()
 
+    def __iter__(self):
+        while True:
+            yield self.read()
+
     def __enter__(self):
         return self
 

linuxpy/video/raw.py

@@ -966,6 +966,18 @@ class TimeCodeFlag(enum.IntFlag):
     COLORFRAME = 0x2
 
 
+class EventSubscriptionFlag(enum.IntFlag):
+    SEND_INITIAL = 1 << 0
+    ALLOW_FEEDBACK = 1 << 1
+
+
+class EventControlChange(enum.IntEnum):
+    VALUE = 1 << 0
+    FLAGS = 1 << 1
+    RANGE = 1 << 2
+    DIMENSIONS = 1 << 3
+
+
 class EventType(enum.IntEnum):
     ALL = 0x0
     VSYNC = 0x1
@@ -974,18 +986,12 @@ class EventType(enum.IntEnum):
     FRAME_SYNC = 0x4
     SOURCE_CHANGE = 0x5
     MOTION_DET = 0x6
-    CTRL_CH_VALUE = 1 << 0
-    CTRL_CH_FLAGS = 1 << 1
-    CTRL_CH_RANGE = 1 << 2
-    CTRL_CH_DIMENSIONS = 1 << 3
     SRC_CH_RESOLUTION = 1 << 0
     MD_FL_HAVE_FRAME_SEQ = 1 << 0
-    SUB_FL_SEND_INITIAL = 1 << 0
-    SUB_FL_ALLOW_FEEDBACK = 1 << 1
 
 
-class EventSubscriptionFlag(enum.IntFlag):
-    pass
+class MbusFrameFormatFlag(enum.IntFlag):
+    SET_CSC = 0x1
 
 
 class Interlaced(enum.IntEnum):

tests/test_video.py

@@ -30,6 +30,10 @@
     ControlClass,
     ControlType,
     Device,
+    EventControlChange,
+    EventReader,
+    EventSubscriptionFlag,
+    EventType,
     InputCapabilities,
     Memory,
     MetaFormat,
@@ -1033,3 +1037,42 @@ def _():
         assert StandardID.PAL_B in capture_dev.get_std()
 
         assert StandardID.PAL_B in capture_dev.query_std()
+
+
+@vivid_only
+@test("vivid events")
+def _():
+    with Device(VIVID_CAPTURE_DEVICE) as capture_dev:
+        capture_dev.set_input(0)
+        brightness = capture_dev.controls.brightness
+        capture_dev.subscribe_event(
+            EventType.CTRL, brightness.id, EventSubscriptionFlag.ALLOW_FEEDBACK | EventSubscriptionFlag.SEND_INITIAL
+        )
+        with EventReader(capture_dev) as reader:
+            initial_value = brightness.value
+            stream = iter(reader)
+            event = next(stream)
+            assert event.u.ctrl.value == initial_value
+            new_value = initial_value + 1 if initial_value < brightness.maximum else initial_value - 1
+            brightness.value = new_value
+            for event in reader:
+                assert event.type == EventType.CTRL
+                assert event.u.ctrl.value == new_value
+                assert event.u.ctrl.changes == EventControlChange.VALUE
+                break
+
+
+@vivid_only
+@test("async vivid events")
+async def _():
+    with Device(VIVID_CAPTURE_DEVICE) as capture_dev:
+        capture_dev.set_input(0)
+        brightness = capture_dev.controls.brightness
+        capture_dev.subscribe_event(EventType.CTRL, brightness.id, EventSubscriptionFlag.ALLOW_FEEDBACK)
+        async with EventReader(capture_dev) as reader:
+            value = brightness.value
+            brightness.value = value + 1 if value < brightness.maximum else value - 1
+            async for event in reader:
+                assert event.type == EventType.CTRL
+                assert event.u.ctrl.changes == EventControlChange.VALUE
+                break