fix(camera): add optional pix_fmt for hardware encoders

bot/camera.py

@@ -89,6 +89,7 @@ def __init__(self, config: ConfigWrapper, klippy: Klippy, logging_handler: loggi
         self._flip_vertically: bool = config.camera.flip_vertically
         self._flip_horizontally: bool = config.camera.flip_horizontally
         self._fourcc: str = config.camera.fourcc
+        self._pix_fmt: str = config.camera.pix_fmt
         self._video_duration: int = config.camera.video_duration
         self._video_buffer_size: int = config.camera.video_buffer_size
         self._stream_fps: int = config.camera.stream_fps
@@ -406,6 +407,8 @@ def process_video_frame(frame_local):
                 codec=self._fourcc,
                 fps=res_fps,
             )
+            if self._pix_fmt:
+                out.target_pix_fmt = self._pix_fmt
 
             for el in frame_list:
                 loc_loc = pickle.loads(el)
@@ -536,6 +539,8 @@ def _create_timelapse(self, printing_filename: str, gcode_name: str, info_mess:
                 codec=self._fourcc,
                 fps=lapse_fps,
             )
+            if self._pix_fmt:
+                out.target_pix_fmt = self._pix_fmt
 
             asyncio.run_coroutine_threadsafe(info_mess.edit_text(text="Images recoding"), loop).result()
             last_update_time = time.time()
@@ -771,6 +776,8 @@ def take_video(self) -> Tuple[BytesIO, BytesIO, int, int]:
                 codec=self._fourcc,
                 fps=res_fps,
             )
+            if self._pix_fmt:
+                out.target_pix_fmt = self._pix_fmt
 
             for el in frame_list:
                 loc_loc = pickle.loads(el)

bot/configuration.py

@@ -255,6 +255,7 @@ class CameraConfig(ConfigHelper):
         "flip_horizontally",
         "rotate",
         "fourcc",
+        "pix_fmt",
         "video_duration",
         "video_buffer_size",
         "fps",
@@ -273,7 +274,8 @@ def __init__(self, config: configparser.ConfigParser):
         self.flip_vertically: bool = self._get_boolean("flip_vertically", default=False)
         self.flip_horizontally: bool = self._get_boolean("flip_horizontally", default=False)
         self.rotate: str = self._get_str("rotate", default="", allowed_values=["", "90_cw", "90_ccw", "180"])
-        self.fourcc: str = self._get_str("fourcc", default="h264", allowed_values=["h264", "mpeg4"])
+        self.fourcc: str = self._get_str("fourcc", default="h264", allowed_values=["h264", "mpeg4", "h264_rkmpp"])
+        self.pix_fmt: str = self._get_str("pix_fmt", default="")
 
         # self.threads: int = self._getint( "threads", fallback=int(len(os.sched_getaffinity(0)) / 2)) #Fixme:
         self.threads: int = self._get_int("threads", default=2, min_value=0)  # Fixme: fix default calcs! add check max value cpu count

tests/configuration_test.py

@@ -46,3 +46,27 @@ def test_config_has_no_errors(config_helper):
 
 def test_config_bot_is_valid(config_helper):
     assert config_helper.secrets.chat_id == 16612341234 and config_helper.secrets.token == "23423423334:sdfgsdfg-dfgdfgsdfg"
+
+
+def test_camera_pix_fmt_defaults_to_empty(config_helper):
+    assert config_helper.camera.pix_fmt == ""
+
+
+def test_camera_pix_fmt_accepts_value(tmp_path):
+    cfg = tmp_path / "telegram.conf"
+    cfg.write_text(
+        "[bot]\nbot_token: x\nchat_id: 1\n[camera]\nhost: http://localhost:8080/?action=stream\ntype: mjpeg\npix_fmt: nv12\n"
+    )
+    wrapper = ConfigWrapper(cfg.as_posix())
+    assert wrapper.configuration_errors == ""
+    assert wrapper.camera.pix_fmt == "nv12"
+
+
+def test_camera_fourcc_accepts_rkmpp(tmp_path):
+    cfg = tmp_path / "telegram.conf"
+    cfg.write_text(
+        "[bot]\nbot_token: x\nchat_id: 1\n[camera]\nhost: http://localhost:8080/?action=stream\ntype: mjpeg\nfourcc: h264_rkmpp\n"
+    )
+    wrapper = ConfigWrapper(cfg.as_posix())
+    assert wrapper.configuration_errors == ""
+    assert wrapper.camera.fourcc == "h264_rkmpp"