I/O helpers

softlora.io

IQ file loading and packet persistence helpers.

iter_iq_chunks(path, chunk_samples=_DEFAULT_CHUNK_SAMPLES)

Stream a recording as complex64 chunks without loading it fully.

Yields complex64 chunks of up to chunk_samples samples from a raw interleaved IQ file (float32 or int16, auto-detected) or a stereo WAV (int16 PCM or float32 IEEE). The first chunk is a true prefix of the stream, so concatenating the chunks reproduces :func:load_iq exactly.

Parameters:

Name	Type	Description	Default
path	str or PathLike	Recording to read.	required
chunk_samples	int	Maximum number of complex samples per yielded chunk.	_DEFAULT_CHUNK_SAMPLES

Yields:

Type	Description
ndarray	Complex baseband samples (complex64).

Source code in softlora/io.py

def iter_iq_chunks(path, chunk_samples=_DEFAULT_CHUNK_SAMPLES):
    """Stream a recording as complex64 chunks without loading it fully.

    Yields ``complex64`` chunks of up to ``chunk_samples`` samples from a raw
    interleaved IQ file (float32 or int16, auto-detected) or a stereo WAV
    (int16 PCM or float32 IEEE).  The first chunk is a true prefix of the
    stream, so concatenating the chunks reproduces :func:`load_iq` exactly.

    Parameters
    ----------
    path : str or os.PathLike
        Recording to read.
    chunk_samples : int
        Maximum number of complex samples per yielded chunk.

    Yields
    ------
    np.ndarray
        Complex baseband samples (complex64).
    """
    path = os.fspath(path)
    ext = os.path.splitext(path)[1].lower()
    if ext in _WAV_EXTENSIONS:
        yield from _iter_wav_chunks(path, chunk_samples)
    elif ext in _RAW_EXTENSIONS or ext == '':
        yield from _iter_raw_chunks(path, chunk_samples)
    else:
        raise ValueError(f'Unrecognized IQ file format: {path!r}')

load_iq(path)

Load a complex baseband IQ recording.

Supported formats (auto-detected):

• Stereo WAV (.wav) -- channel 0 = I, channel 1 = Q (int16 or float32)
• Raw interleaved I/Q float32 (.cfile, GNU Radio File Sink output)
• Raw interleaved I/Q int16 (.dat) -- detected via a power check

Returns:

Type	Description
ndarray	Complex baseband samples (complex64).

Source code in softlora/io.py

def load_iq(path):
    """Load a complex baseband IQ recording.

    Supported formats (auto-detected):

    * Stereo WAV (``.wav``) -- channel 0 = I, channel 1 = Q (int16 or float32)
    * Raw interleaved I/Q float32 (``.cfile``, GNU Radio File Sink output)
    * Raw interleaved I/Q int16 (``.dat``) -- detected via a power check

    Returns
    -------
    np.ndarray
        Complex baseband samples (complex64).
    """
    path = os.fspath(path)
    ext = os.path.splitext(path)[1].lower()

    if ext in _WAV_EXTENSIONS:
        from scipy.io import wavfile
        rate, data = wavfile.read(path)
        del rate
        if data.ndim < 2:
            raise ValueError(
                f'{path}: mono WAV has no I/Q channels; need a stereo recording'
            )
        i = data[:, 0]
        q = data[:, 1]
        if i.dtype.kind == 'i':  # int16 and similar -> [-1, 1)
            i = i.astype(np.float32) / 32768.0
            q = q.astype(np.float32) / 32768.0
        return (i + 1j * q).astype(np.complex64)

    if ext in _RAW_EXTENSIONS or ext == '':
        size = os.path.getsize(path)
        if size % 8 != 0:
            # size is odd in complex64 units -> not interleaved float32
            pass
        data = np.fromfile(path, dtype=np.float32)
        if data.size % 2:
            data = data[:-1]
        iq = (data[0::2] + 1j * data[1::2]).astype(np.complex64)
        if np.mean(np.abs(iq[:1000]) ** 2) > 1e-6:
            return iq
        data = np.fromfile(path, dtype=np.int16)
        if data.size % 2:
            data = data[:-1]
        i = data[0::2].astype(np.float32) / 32768.0
        q = data[1::2].astype(np.float32) / 32768.0
        return (i + 1j * q).astype(np.complex64)

    raise ValueError(f'Unrecognized IQ file format: {path!r}')

sample_rate(path)

Sample rate stored in a WAV file header, or None for raw IQ files.

Source code in softlora/io.py

def sample_rate(path):
    """Sample rate stored in a WAV file header, or None for raw IQ files."""
    if os.path.splitext(path)[1].lower() in _WAV_EXTENSIONS:
        from scipy.io import wavfile
        return int(wavfile.read(path)[0])
    return None

save_packets(packets, out_dir)

Persist decoded packets.

Writes, per packet, one raw payload packet_XXXX.bin plus an append-only metadata.jsonl line with each packet's :meth:Packet.to_dict serialization. An append-only packets.txt summary renders each packet as its human-readable :meth:Packet.__str__ detail table for quick inspection.

Parameters:

Name	Type	Description	Default
packets	list of Packet	Packets to save.	required
out_dir	str or PathLike	Directory to create and write into.	required

Returns:

Type	Description
list of str	Absolute paths written.

Source code in softlora/io.py

def save_packets(packets, out_dir):
    """Persist decoded packets.

    Writes, per packet, one raw payload ``packet_XXXX.bin`` plus an append-only
    ``metadata.jsonl`` line with each packet's :meth:`Packet.to_dict`
    serialization.  An append-only ``packets.txt`` summary renders each packet
    as its human-readable :meth:`Packet.__str__` detail table for quick
    inspection.

    Parameters
    ----------
    packets : list of Packet
        Packets to save.
    out_dir : str or os.PathLike
        Directory to create and write into.

    Returns
    -------
    list of str
        Absolute paths written.
    """
    os.makedirs(out_dir, exist_ok=True)
    meta_path = os.path.join(out_dir, 'metadata.jsonl')
    summary_path = os.path.join(out_dir, 'packets.txt')
    written = []
    with open(meta_path, 'a', encoding='utf-8') as fh, \
            open(summary_path, 'a', encoding='utf-8') as sh:
        for p in packets:
            idx = p.packet_index if p.packet_index is not None else len(written)
            name = f'packet_{idx:04d}.bin'
            payload_path = os.path.join(out_dir, name)
            with open(payload_path, 'wb') as pf:
                pf.write(np.asarray(p.payload_bytes, dtype=np.uint8).tobytes())
            written.append(payload_path)
            fh.write(json.dumps(p.to_dict()) + '\n')
            sh.write(f'== {name} ==\n')
            sh.write(str(p))
            sh.write('\n\n')
    return written