iqfile example¶

import os
from iqfile.conv import Q15ToCf32

fd = os.open("samples.q15", os.O_RDONLY | getattr(os, "O_BINARY", 0))
reader = Q15ToCf32(fd=fd)
block  = reader.steps(1024)   # returns complex64 ndarray
os.close(fd)

"""Implement iqfile_cf32_to_q15_step() and add the samples_written counter."""

from pathlib import Path
import sys

root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()

# ── step() in _core.h ──────────────────────────────────────────────────────
core_h = root / "native/inc/iqfile/cf32_to_q15/cf32_to_q15_core.h"
text = core_h.read_text(encoding="utf-8")

if "<math.h>" not in text:
    text = text.replace(
        '#include "iqfile/clib_common.h"',
        '#include "iqfile/clib_common.h"\n#include <math.h>',
        1,
    )

OLD = """\
    (void)state; /* TODO: implement using state variables */
    return (int32_t)x;"""

NEW = """\
    float s = state->scale;
    int16_t i = (int16_t)fmaxf(-s, fminf(s, crealf(x) * s));
    int16_t q = (int16_t)fmaxf(-s, fminf(s, cimagf(x) * s));
    /* Pack I in low 16 bits, Q in high 16 bits.
     * Python: packed.view(np.int16) gives interleaved [i0, q0, i1, q1, ...] */
    return (int32_t)((uint32_t)(uint16_t)i | ((uint32_t)(uint16_t)q << 16));"""

assert OLD in text, "step stub not found — was it already patched?"
core_h.write_text(text.replace(OLD, NEW, 1), encoding="utf-8")
print(f"patched  {core_h.relative_to(root)}")

# ── samples_written counter in _core.c ────────────────────────────────────
core_c = root / "native/src/cf32_to_q15/cf32_to_q15_core.c"
text = core_c.read_text(encoding="utf-8")

OLD_LOOP = """\
    for (size_t i = 0; i < n; i++)
        output[i] = iqfile_cf32_to_q15_step(state, input[i]);
}"""

NEW_LOOP = """\
    for (size_t i = 0; i < n; i++)
        output[i] = iqfile_cf32_to_q15_step(state, input[i]);
    state->samples_written += (uint32_t)n;
}"""

assert OLD_LOOP in text, "steps() loop not found"
core_c.write_text(text.replace(OLD_LOOP, NEW_LOOP, 1), encoding="utf-8")
print(f"patched  {core_c.relative_to(root)}")

"""Implement iqfile_q15_to_cf32_step(), samples_read counter, and eof getter."""

from pathlib import Path
import sys

root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()

# read()/lseek() live in <unistd.h> on POSIX and <io.h> on Windows, under the
# same names -- so one include block, used by both files below.
POSIX_IO = """\
#include <stdio.h>     /* SEEK_SET/CUR/END: <io.h> lacks them */
#include <sys/types.h> /* off_t */
#ifdef _WIN32
#include <io.h> /* read, lseek: the POSIX names, as the CRT spells them */
#else
#include <unistd.h>
#endif"""

# ── add the I/O header to _core.h ─────────────────────────────────────────
core_h = root / "native/inc/iqfile/q15_to_cf32/q15_to_cf32_core.h"
text = core_h.read_text(encoding="utf-8")

if "<unistd.h>" not in text:
    text = text.replace(
        '#include "iqfile/clib_common.h"',
        '#include "iqfile/clib_common.h"\n' + POSIX_IO,
        1,
    )

# ── step() in _core.h ──────────────────────────────────────────────────────
OLD = """\
    (void)state; /* TODO: implement */
    return (float _Complex)0;"""

NEW = """\
    int16_t pair[2] = {0, 0};
    if (state->fd >= 0)
        read((int)state->fd, pair, sizeof(pair));
    return ((float)pair[0] + (float)pair[1] * I) / state->scale;"""

assert OLD in text, "step stub not found — was it already patched?"
text = text.replace(OLD, NEW, 1)

# Add eof getter declaration before the closing header guard #endif
guard = "#endif /* IQFILE_Q15_TO_CF32_CORE_H */"
assert guard in text, "header guard not found"
text = text.replace(
    guard,
    "int32_t iqfile_q15_to_cf32_get_eof(const iqfile_q15_to_cf32_state_t *state);\n\n"
    + guard,
    1,
)
core_h.write_text(text, encoding="utf-8")
print(f"patched  {core_h.relative_to(root)}")

# ── samples_read counter + eof getter in _core.c ──────────────────────────
core_c = root / "native/src/q15_to_cf32/q15_to_cf32_core.c"
text = core_c.read_text(encoding="utf-8")

# The I/O header if needed (steps() calls read/lseek)
if "<unistd.h>" not in text:
    text = text.replace(
        '#include "iqfile/q15_to_cf32/q15_to_cf32_core.h"',
        '#include "iqfile/q15_to_cf32/q15_to_cf32_core.h"\n' + POSIX_IO,
        1,
    )

# Counter in steps()
OLD_LOOP = """\
    for (size_t i = 0; i < n; i++)
        output[i] = iqfile_q15_to_cf32_step(state);
}"""

NEW_LOOP = """\
    for (size_t i = 0; i < n; i++)
        output[i] = iqfile_q15_to_cf32_step(state);
    state->samples_read += (uint32_t)n;
}"""

assert OLD_LOOP in text, "steps() loop not found"
text = text.replace(OLD_LOOP, NEW_LOOP, 1)

# eof getter: `jm property` scaffolded a marked placeholder for it (step 3);
# fill that in rather than writing a second definition beside it.
OLD_EOF = """\
/* <<IMPLEMENT: iqfile_q15_to_cf32_get_eof>> */
int32_t
iqfile_q15_to_cf32_get_eof(const iqfile_q15_to_cf32_state_t *state)
{
    (void)state;
    return 0; /* placeholder */
}"""

NEW_EOF = """\
int32_t
iqfile_q15_to_cf32_get_eof(const iqfile_q15_to_cf32_state_t *state)
{
    if (state->fd < 0)
        return 1;
    off_t cur = lseek((int)state->fd, 0, SEEK_CUR);
    off_t end = lseek((int)state->fd, 0, SEEK_END);
    lseek((int)state->fd, cur, SEEK_SET);
    return cur == end ? 1 : 0;
}"""

assert OLD_EOF in text, "eof placeholder not found — was it already patched?"
text = text.replace(OLD_EOF, NEW_EOF, 1)
core_c.write_text(text, encoding="utf-8")
print(f"patched  {core_c.relative_to(root)}")

packed = writer.steps(cf32_block)          # dtype int32, shape (N,)
q15    = packed.view(np.int16)             # dtype int16, shape (2N,) — [i0,q0,i1,q1,…]
q15.tofile("samples.q15")

import os
fd     = os.open("samples.q15", os.O_RDONLY | getattr(os, "O_BINARY", 0))
reader = Q15ToCf32(fd=fd)
block  = reader.steps(1024)    # reads 4 KiB, returns complex64 ndarray
os.close(fd)

@final
class Q15ToCf32:
    """Read interleaved q15 (int16 I/Q) samples as complex float.
    ...
    """
    @property
    def eof(self) -> int:
        """True (1) once the backing file descriptor is exhausted."""

"""Enrich the sacred ``<obj>_core.h`` headers with Doxygen so the generated
``conv.pyi`` carries real class summaries and rich property docstrings.

The header is the single source of truth for documentation: ``jm`` parses these
``/** ... */`` comments and turns them into numpy-style Python docstrings. Run
this after the C kernels are implemented (``04_patch_writer.py`` /
``04_patch_reader.py``); a follow-up ``jm apply`` re-derives the glue (``.pyi``
included) from the edited headers.

Which source flows where — verified empirically (mirrors the views_module
finding):

  - **Class summary** comes from ``<obj>_create``'s ``@brief``. Both converter
    objects own a real ``<obj>_create`` (they are distinct module objects, not
    views), so a header ``@brief`` enriches each class summary.
  - **Field-backed** property docstrings (``samples_written``, ``samples_read``)
    come from the manifest ``jm property --doc`` value, NOT a header ``@brief``:
    a ``--field`` getter is auto-implemented inline, so there is no header
    declaration to annotate. Those docs are set in ``test.py`` / step 3.
  - A **computed** property's getter has a real declaration, so it CAN carry a
    header ``@brief``. ``eof`` is computed (``iqfile_q15_to_cf32_get_eof``), so its
    docstring is enriched here.

iqfile exposes no named ``jm method`` (only ``step``/``steps`` plus
properties), and a property getter's docstring renders as prose only — a
``@code`` block on a property is not turned into a runnable doctest. So this
example ships rich class summaries and property docs, but no runnable method
doctest (like the other "light" examples).

Usage:  python3 04b_doxygen.py     # saved in, and run from, the project root
"""

from __future__ import annotations

import pathlib
import re

from just_makeit import _csym  # gh-1591: the derived symbol stem
import sys

# Class summaries, injected over jm's scaffold brief on each <obj>_create.
CREATE_BRIEFS = {
    "cf32_to_q15": (
        "Pack complex float samples into interleaved q15 (int16 I/Q)."
    ),
    "q15_to_cf32": (
        "Read interleaved q15 (int16 I/Q) samples as complex float."
    ),
}

# Computed-property getters keyed by the C declaration they sit above. Their
# @brief becomes the property __doc__ (field-backed props are documented via
# the manifest instead — see the module docstring).
GETTER_BLOCKS = {
    "q15_to_cf32": [
        (
            "int32_t iqfile_q15_to_cf32_get_eof(",
            "/**\n"
            " * @brief True (1) once the backing file descriptor is"
            " exhausted.\n"
            " */\n",
        ),
    ],
}


def _enrich(obj: str) -> None:
    header = pathlib.Path("native/inc/iqfile") / obj / f"{obj}_core.h"
    text = header.read_text(encoding="utf-8")

    # Replace jm's trivial scaffold brief on <obj>_create with a real summary.
    scaffold_re = re.compile(
        rf"/\*\*\n \* @brief Create a {obj} instance\..*?"
        rf"(?={_csym.stem(header, obj)}_state_t \*{_csym.stem(header, obj)}_create)",
        re.DOTALL,
    )
    new_create = f"/**\n * @brief {CREATE_BRIEFS[obj]}\n */\n"
    text, n = scaffold_re.subn(new_create, text, count=1)
    if n != 1:
        print(
            f"ERROR: {obj} create() scaffold brief not found", file=sys.stderr
        )
        sys.exit(1)

    # Prepend each computed getter's Doxygen block above its declaration.
    for decl, block in GETTER_BLOCKS.get(obj, []):
        if decl not in text:
            print(f"ERROR: declaration not found: {decl!r}", file=sys.stderr)
            sys.exit(1)
        text = text.replace(decl, block + decl, 1)

    header.write_text(text, encoding="utf-8")
    print(f"enriched {header}")


def main() -> None:
    for obj in CREATE_BRIEFS:
        _enrich(obj)


if __name__ == "__main__":
    main()

"""Round-trip demo: cf32 -> q15 file -> cf32, verify fidelity."""

import os
import sys
import tempfile
import numpy as np

# cwd is the project root when called from test.py; cmake builds the .so into src/.
sys.path.insert(0, "src")

from iqfile.conv import Cf32ToQ15, Q15ToCf32

N = 4096
rng = np.random.default_rng(42)

# ── Generate test signal (normalised to [-0.9, 0.9] to avoid clipping) ────
signal = (rng.standard_normal(N) + 1j * rng.standard_normal(N)).astype(
    np.complex64
)
signal *= 0.9 / np.max(np.abs(signal))

# ── Write cf32 -> q15 ─────────────────────────────────────────────────────
writer = Cf32ToQ15()
packed = writer.steps(signal)  # int32 array, shape (N,)
q15 = packed.view(np.int16)  # int16 view, shape (2N,)

with tempfile.NamedTemporaryFile(suffix=".q15", delete=False) as f:
    q15_path = f.name
    q15.tofile(f)

print(
    f"wrote    {N} complex samples -> {q15_path}  ({os.path.getsize(q15_path)} bytes)"
)
print(f"written: {writer.samples_written} samples")

# ── Read q15 -> cf32 ──────────────────────────────────────────────────────
fd = os.open(q15_path, os.O_RDONLY | getattr(os, "O_BINARY", 0))
reader = Q15ToCf32(fd=fd)
recovered = reader.steps(N)

print(f"read:    {reader.samples_read} samples,  eof={reader.eof}")
os.close(fd)

# ── Verify round-trip fidelity ────────────────────────────────────────────
scale = 32767.0
quantisation_noise_floor = 1.0 / scale  # ≈ -90 dB
err = np.max(np.abs(signal - recovered))

print(f"max err: {err:.6f}  (floor ~{quantisation_noise_floor:.6f})")
assert err < 2.0 / scale, f"round-trip error too large: {err}"
print("PASSED")

os.unlink(q15_path)