Merge branch 'main' into fix-186-avoid-decode

pyproject.toml

@@ -4,8 +4,8 @@ version = "0.1.0"
 description = ""
 authors = ["lucemia <lucemia@gmail.com>"]
 readme = "README.md"
-packages = [{ include = "ffmpeg", from = "src" }]
-include = ["ffmpeg/py.typed"]
+packages = [{ include = "typed_ffmpeg", from = "src" }]
+include = ["typed_ffmpeg/py.typed"]
 exclude = ["**/tests"]
 
 [tool.poetry.dependencies]

scripts/requirements.txt

@@ -29,12 +29,12 @@ multidict==6.0.4
 mypy-extensions==1.0.0
 mypy==1.8.0
 nodeenv==1.8.0
-openai==1.6.1
+openai==1.7.2
 packaging==23.2
 platformdirs==4.1.0
 pluggy==1.3.0
 pre-commit==3.6.0
-pydantic-core==2.14.6
+pydantic-core==2.15.0
 pydantic-settings==2.1.0
 pydantic==2.5.3
 pygments==2.17.2

src/typed_ffmpeg/__init__.py

@@ -0,0 +1,6 @@
+from . import filters, nodes, streams
+from .base import input, merge_outputs, output
+from .probe import probe
+
+__all__ = ["input", "output", "merge_outputs", "probe", "filters"]
+__all__ += streams.__all__ + nodes.__all__

src/typed_ffmpeg/nodes/__init__.py

@@ -0,0 +1,3 @@
+from .base import Stream
+
+__all__ = ["Stream"]

src/typed_ffmpeg/probe.py

@@ -0,0 +1,33 @@
+import json
+import subprocess
+from typing import Any
+
+from .utils.escaping import convert_kwargs_to_cmd_line_args
+
+
+def probe(filename: str, cmd: str = "ffprobe", timeout: int | None = None, **kwargs: Any) -> dict[str, Any]:
+    """Run ffprobe on the specified file and return a JSON representation of the output.
+
+    Raises:
+        :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code,
+            an :class:`Error` is returned with a generic error message.
+            The stderr output can be retrieved by accessing the
+            ``stderr`` property of the exception.
+    """
+    args = [cmd, "-show_format", "-show_streams", "-of", "json"]
+    args += convert_kwargs_to_cmd_line_args(kwargs)
+    args += [filename]
+
+    p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+    if timeout is not None:
+        out, err = p.communicate(timeout=timeout)
+    else:
+        out, err = p.communicate()
+
+    if p.returncode != 0:
+        raise subprocess.SubprocessError("ffprobe", out, err)
+    return json.loads(out.decode("utf-8"))
+
+
+__all__ = ["probe"]

src/typed_ffmpeg/streams/__init__.py

@@ -0,0 +1,5 @@
+from .audio import AudioStream
+from .av import AVStream
+from .video import VideoStream
+
+__all__ = ["AudioStream", "VideoStream", "AVStream"]

src/typed_ffmpeg/utils/dataclasses_helper.py

@@ -0,0 +1,46 @@
+import json
+from dataclasses import asdict, is_dataclass
+from typing import Any, ClassVar, Dict, Protocol, TypeVar, cast
+
+
+class IsDataclass(Protocol):
+    # as already noted in comments, checking for this attribute is currently
+    # the most reliable way to ascertain that something is a dataclass
+    __dataclass_fields__: ClassVar[Dict[str, Any]]
+
+
+T = TypeVar("T", bound=IsDataclass)
+
+
+def load(cls: type[T], raw: str) -> T:
+    data = json.loads(raw)
+    return _load(cls, data)
+
+
+def _load(cls: type[T], data: dict[str, Any]) -> T:
+    """
+    Deserialize a dictionary into a dataclass of type cls.
+
+    :param cls: The type of the dataclass to deserialize into.
+    :param data: A dictionary representing the serialized dataclass.
+    :return: An instance of the specified dataclass.
+    """
+
+    if is_dataclass(cls):
+        field_types = cls.__annotations__
+        field_data = {}
+        for field_name, field_type in field_types.items():
+            field_value = data.get(field_name)
+            if is_dataclass(field_type) and isinstance(field_value, dict):
+                # Recursive call for nested dataclasses
+                field_data[field_name] = _load(field_type, field_value)
+            else:
+                field_data[field_name] = field_value
+        return cast(T, cls(**field_data))
+
+    raise TypeError(f"The provided class {cls.__name__} is not a dataclass")
+
+
+# Serialization
+def dump(instance: IsDataclass) -> str:
+    return json.dumps(asdict(instance))

src/typed_ffmpeg/utils/escaping.py

@@ -0,0 +1,32 @@
+from typing import Any, Iterable
+
+
+def escape(text: str | int | tuple[int, int], chars: str = "\\'=:") -> str:
+    """Helper function to escape uncomfortable characters."""
+    text = str(text)
+    _chars = list(set(chars))
+    if "\\" in _chars:
+        _chars.remove("\\")
+        _chars.insert(0, "\\")
+
+    for ch in _chars:
+        text = text.replace(ch, "\\" + ch)
+
+    return text
+
+
+def convert_kwargs_to_cmd_line_args(kwargs: dict[str, Any]) -> list[str]:
+    """Helper function to build command line arguments out of dict."""
+    args = []
+    for k in sorted(kwargs.keys()):
+        v = kwargs[k]
+        if isinstance(v, Iterable) and not isinstance(v, str):
+            for value in v:
+                args.append("-{}".format(k))
+                if value is not None:
+                    args.append("{}".format(value))
+            continue
+        args.append("-{}".format(k))
+        if v is not None:
+            args.append("{}".format(v))
+    return args

src/typed_ffmpeg/utils/tests/__snapshots__/test_dataclasses_helpers.ambr

@@ -0,0 +1,7 @@
+# serializer version: 1
+# name: test_load_and_dump[deserialized]
+  Person(name='John Doe', age=30, address=Address(street='123 Main St', city='Anytown'))
+# ---
+# name: test_load_and_dump[serialized]
+  '{"name": "John Doe", "age": 30, "address": {"street": "123 Main St", "city": "Anytown"}}'
+# ---

src/typed_ffmpeg/utils/tests/test_dataclasses_helpers.py

@@ -0,0 +1,32 @@
+from dataclasses import dataclass
+
+from syrupy.assertion import SnapshotAssertion
+
+from ..dataclasses_helper import dump, load
+
+
+# Define your dataclasses
+@dataclass
+class Address:
+    street: str
+    city: str
+
+
+@dataclass
+class Person:
+    name: str
+    age: int
+    address: Address
+
+
+def test_load_and_dump(snapshot: SnapshotAssertion) -> None:
+    # Example usage
+    person = Person("John Doe", 30, Address("123 Main St", "Anytown"))
+    serialized = dump(person)
+    assert snapshot(name="serialized") == serialized
+
+    deserialized = load(Person, serialized)
+
+    assert isinstance(deserialized, Person)
+    assert snapshot(name="deserialized") == deserialized
+    assert person == deserialized