Merge pull request #1 from vkottler/dev/0.2.0

0.2.0 - Initial file interactions

.github/workflows/python-package.yml

@@ -68,7 +68,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=quasimoto version=0.1.0
+                            repo=quasimoto version=0.2.0
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

.gitignore

@@ -15,3 +15,4 @@ coverage*.xml
 tags
 mklocal
 docs
+test.wav

.pylintrc

@@ -0,0 +1,2 @@
+[DESIGN]
+max-args=6

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=32f83dc4656170a9cfc0c52220eb0ead
+    hash=c8fbe39d4a560fc68d4d3bba34f88ccd
     =====================================
 -->
 
-# quasimoto ([0.1.0](https://pypi.org/project/quasimoto/))
+# quasimoto ([0.2.0](https://pypi.org/project/quasimoto/))
 
 [![python](https://img.shields.io/pypi/pyversions/quasimoto.svg)](https://pypi.org/project/quasimoto/)
 ![Build Status](https://github.com/vkottler/quasimoto/workflows/Python%20Package/badge.svg)

local/configs/package.yaml

@@ -5,10 +5,17 @@ description: A lossless audio generator.
 entry: {{entry}}
 requirements:
   - vcorelib
+  - runtimepy
+  - scipy
+  - matplotlib
 dev_requirements:
   - setuptools-wrapper
   - types-setuptools
 
 commands:
   - name: gen
     description: "generate audio"
+
+mypy_local: |
+  [mypy-scipy.*]
+  ignore_missing_imports = True

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 0
-minor: 1
+minor: 2
 patch: 0
 entry: quasimoto

mypy.ini

@@ -13,3 +13,7 @@ warn_unused_ignores = False
 strict = False
 disallow_any_generics = False
 strict_equality = False
+
+# quasimoto-specific configurations.
+[mypy-scipy.*]
+ignore_missing_imports = True

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "quasimoto"
-version = "0.1.0"
+version = "0.2.0"
 description = "A lossless audio generator."
 readme = "README.md"
 requires-python = ">=3.11"

quasimoto/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.4
-# hash=5270e41d1149e6ca425f99c76f92c409
+# hash=8efc396041b8d6d582b53a5468ffda71
 # =====================================
 
 """
@@ -10,4 +10,4 @@
 
 DESCRIPTION = "A lossless audio generator."
 PKG_NAME = "quasimoto"
-VERSION = "0.1.0"
+VERSION = "0.2.0"

quasimoto/commands/gen.py

@@ -4,30 +4,22 @@
 
 # built-in
 import argparse
-from enum import StrEnum
 from pathlib import Path
 
 # third-party
 from vcorelib.args import CommandFunction
 
 # internal
 from quasimoto import PKG_NAME
-
-
-class AudioFileTypes(StrEnum):
-    """An enumeration for supported file types."""
-
-    WAVE = "wav"
-
-
-DEFAULT_FORMAT = AudioFileTypes.WAVE
+from quasimoto.enums import DEFAULT_FORMAT
+from quasimoto.riff import RiffInterface
 
 
 def gen_cmd(args: argparse.Namespace) -> int:
     """Execute the arbiter command."""
 
-    with args.output.open("wb") as out_fd:
-        print(out_fd)
+    with RiffInterface.from_path(args.output) as writer:
+        print(writer)
 
     return 0
 

quasimoto/enums/__init__.py

@@ -0,0 +1,55 @@
+"""
+A module implementing enumeration interfaces for this package.
+"""
+
+# built-in
+from enum import StrEnum
+from typing import BinaryIO, Optional
+
+
+class AudioFileTypes(StrEnum):
+    """An enumeration for supported file types."""
+
+    WAVE = "wav"
+
+
+DEFAULT_FORMAT = AudioFileTypes.WAVE
+
+
+class ChunkType(StrEnum):
+    """An enumeration for different kinds of RIFF chunks."""
+
+    RIFF = "RIFF"
+    LIST = "LIST"
+    WAVE = "WAVE"
+
+    FMT = "fmt "
+    DATA = "data"
+    ID3 = "ID3 "
+
+    @property
+    def is_container(self) -> bool:
+        """Whether or not this is a container chunk type."""
+
+        return self is ChunkType.RIFF or self is ChunkType.LIST
+
+    @staticmethod
+    def from_stream(stream: BinaryIO) -> Optional["ChunkType"]:
+        """Read the chunk type from a stream."""
+
+        result = None
+
+        check = stream.read(3).decode("ascii")
+        if len(check) == 3:
+            # Some files hackily have some 'ID3' metadata at the end?
+            if check != "ID3":
+                result = ChunkType(check + stream.read(1).decode("ascii"))
+
+        return result
+
+    def to_stream(self, stream: BinaryIO) -> None:
+        """Write the chunk header."""
+
+        data = bytes(str(self).encode("ascii"))
+        assert len(data) == 4
+        stream.write(data)

quasimoto/requirements.txt

@@ -1 +1,4 @@
 vcorelib
+runtimepy
+scipy
+matplotlib

quasimoto/riff/__init__.py

@@ -0,0 +1,152 @@
+"""
+A module implementing interfaces for RIFF files.
+"""
+
+# built-in
+from contextlib import contextmanager
+import os
+from pathlib import Path
+from typing import BinaryIO, Iterator, Optional, Type, TypeVar, cast
+
+# third-party
+from runtimepy.primitives import Uint32
+from runtimepy.primitives.byte_order import ByteOrder
+from vcorelib.logging import LoggerMixin
+
+# internal
+from quasimoto.enums import ChunkType
+from quasimoto.riff.chunk import NULL_BYTE, Chunk
+
+T = TypeVar("T", bound="RiffInterface")
+
+
+class RiffInterface(LoggerMixin):
+    """A class for reading and writing RIFF files."""
+
+    def __init__(self, stream: BinaryIO, is_writer: bool = True) -> None:
+        """Initialize this instance."""
+
+        super().__init__()
+
+        self.stream = stream
+
+        # Write the header.
+        self.is_writer = is_writer
+        if self.is_writer:
+            ChunkType.RIFF.to_stream(self.stream)
+            # Leave a placeholder for actual size.
+            self.write_size(0)
+        else:
+            header = self.read()
+            assert header is not None
+            self.header: Chunk = header
+            self.logger.info("Header: %s.", self.header)
+            assert self.header.kind is ChunkType.RIFF
+
+    def read_size(self) -> int:
+        """Read a size from the stream."""
+
+        return cast(
+            int,
+            Uint32.kind.read(self.stream, byte_order=ByteOrder.LITTLE_ENDIAN),
+        )
+
+    def read(self) -> Optional[Chunk]:
+        """Read the next chunk."""
+
+        result = None
+
+        print(self.stream.tell())
+        kind = ChunkType.from_stream(self.stream)
+        if kind is not None:
+            size = self.read_size()
+            data = None
+            form = None
+
+            if not kind.is_container:
+                data = self.stream.read(size)
+                if size % 2 == 1:
+                    self.stream.read(1)  # pragma: nocover
+            else:
+                form = ChunkType.from_stream(self.stream)
+
+            result = Chunk(kind, size, data=data, form=form)
+
+        return result
+
+    def chunks(self) -> Iterator[Chunk]:
+        """Read file chunks."""
+
+        result = self.read()
+        while result is not None:
+            yield result
+            result = self.read()
+
+    def write_size(self, size: int, seek: int = None) -> None:
+        """An interface for writing a size field."""
+
+        # Validate size.
+        prim = Uint32.kind
+        bounds = prim.int_bounds
+        assert bounds is not None
+        assert bounds.validate(size), size
+
+        if seek is not None:
+            self.stream.seek(seek)
+
+        # Write size.
+        Uint32.kind.write(
+            size, self.stream, byte_order=ByteOrder.LITTLE_ENDIAN
+        )
+
+    def _write_data(self, data: bytes) -> None:
+        """Write chunk data."""
+
+        size = len(data)
+        self.write_size(size)
+
+        # Write data.
+        self.stream.write(data)
+        if size % 2 == 1:
+            self.stream.write(NULL_BYTE)  # pragma: nocover
+
+    def write(self, chunk: Chunk) -> None:
+        """Write a chunk to the file."""
+
+        assert self.is_writer
+
+        # Can't write container chunks this way.
+        assert not chunk.kind.is_container
+
+        # Write header.
+        chunk.kind.to_stream(self.stream)
+
+        # Write data.
+        if chunk.data is not None:
+            self._write_data(chunk.data)
+
+    def finalize(self) -> None:
+        """Finalize the header size."""
+
+        if self.is_writer:
+            self.stream.seek(0, os.SEEK_END)
+            size = self.stream.tell() - 8
+            self.write_size(size, seek=4)
+        else:
+            remaining = self.stream.read()
+            if remaining:
+                self.logger.warning(
+                    "%d bytes remaining in file!", len(remaining)
+                )
+
+    @classmethod
+    @contextmanager
+    def from_path(
+        cls: Type[T], path: Path, is_writer: bool = True
+    ) -> Iterator[T]:
+        """Create a RIFF interface from a path."""
+
+        with path.open("wb" if is_writer else "rb") as out_fd:
+            result = cls(out_fd, is_writer=is_writer)
+            yield result
+            result.finalize()

quasimoto/riff/chunk.py

@@ -0,0 +1,30 @@
+"""
+A module implementing a RIFF-chunk interface.
+"""
+
+# built-in
+from typing import NamedTuple, Optional
+
+# internal
+from quasimoto.enums import ChunkType
+
+
+class Chunk(NamedTuple):
+    """A container for chunk data."""
+
+    kind: ChunkType
+    size: int
+    data: Optional[bytes] = None
+    form: Optional[ChunkType] = None
+
+    def __str__(self) -> str:
+        """Get this chunk as a string."""
+        result = f"'{self.kind}' size={self.size}"
+
+        if self.form is not None:
+            result += f" (form='{self.form}')"
+
+        return result
+
+
+NULL_BYTE = "\0".encode()