I/O

Readers and writers for tabulated inputs and outputs. Unit conversions belong here, at the I/O boundary — never in the physics or numerics layers.

Readers

liulu.io.readers.read_columns

read_columns(filename, usecols=None, skiprows=0, delimiter=None)

Read columnar data from a text file.

Parameters:

Name	Type	Description	Default
filename	str	Path to file.	required
usecols	tuple of int	Column indices to read.	None
skiprows	int	Number of header rows to skip.	0
delimiter	str	Column delimiter. None = whitespace.	None

Returns:

Type	Description
ndarray	Data array, shape (n_rows, n_cols) or (n_rows,) if single column.

Source code in liulu/io/readers.py

def read_columns(filename, usecols=None, skiprows=0, delimiter=None):
    """Read columnar data from a text file.

    Parameters
    ----------
    filename : str
        Path to file.
    usecols : tuple of int, optional
        Column indices to read.
    skiprows : int
        Number of header rows to skip.
    delimiter : str, optional
        Column delimiter. None = whitespace.

    Returns
    -------
    ndarray
        Data array, shape (n_rows, n_cols) or (n_rows,) if single column.
    """
    return np.loadtxt(filename, usecols=usecols, skiprows=skiprows,
                      delimiter=delimiter)

liulu.io.readers.read_xi_table

read_xi_table(filename, r_col=0, xi_col=1, **kwargs)

Read a tabulated xi(r) file.

Returns:

Name	Type	Description
r	ndarray	Separation, Mpc/h.
xi	ndarray	Correlation function.

Source code in liulu/io/readers.py

def read_xi_table(filename, r_col=0, xi_col=1, **kwargs):
    """Read a tabulated xi(r) file.

    Returns
    -------
    r : ndarray
        Separation, Mpc/h.
    xi : ndarray
        Correlation function.
    """
    data = read_columns(filename, usecols=(r_col, xi_col), **kwargs)
    return data[:, 0], data[:, 1]

liulu.io.readers.read_moments_table

read_moments_table(filename, r_col=0, vr_col=1, sr_col=2, st_col=3, gr_col=None, kr_col=None, kt_col=None, **kwargs)

Read a tabulated pairwise velocity moments file.

Returns:

Type	Description
dict	Keys: 'r', 'v_r', 'sigma_r', 'sigma_t', and optionally 'gamma_r', 'kappa_r', 'kappa_t'.

Source code in liulu/io/readers.py

def read_moments_table(filename, r_col=0, vr_col=1, sr_col=2, st_col=3,
                       gr_col=None, kr_col=None, kt_col=None, **kwargs):
    """Read a tabulated pairwise velocity moments file.

    Returns
    -------
    dict
        Keys: 'r', 'v_r', 'sigma_r', 'sigma_t', and optionally
        'gamma_r', 'kappa_r', 'kappa_t'.
    """
    cols = [r_col, vr_col, sr_col, st_col]
    opt_cols = {'gamma_r': gr_col, 'kappa_r': kr_col, 'kappa_t': kt_col}
    for key, col in opt_cols.items():
        if col is not None:
            cols.append(col)

    data = read_columns(filename, usecols=tuple(cols), **kwargs)

    result = {
        'r': data[:, 0],
        'v_r': data[:, 1],
        'sigma_r': data[:, 2],
        'sigma_t': data[:, 3],
    }

    idx = 4
    for key, col in opt_cols.items():
        if col is not None:
            result[key] = data[:, idx]
            idx += 1

    return result

liulu.io.readers.read_pk_table

read_pk_table(filename, k_col=0, pk_col=1, **kwargs)

Read a tabulated P(k) file.

Returns:

Name	Type	Description
k	ndarray	Wavenumbers, h/Mpc.
pk	ndarray	Power spectrum, (Mpc/h)^3.

Source code in liulu/io/readers.py

def read_pk_table(filename, k_col=0, pk_col=1, **kwargs):
    """Read a tabulated P(k) file.

    Returns
    -------
    k : ndarray
        Wavenumbers, h/Mpc.
    pk : ndarray
        Power spectrum, (Mpc/h)^3.
    """
    data = read_columns(filename, usecols=(k_col, pk_col), **kwargs)
    return data[:, 0], data[:, 1]

Writers

liulu.io.writers.write_xi_2d

write_xi_2d(filename, s_perp, s_par, xi_s, header='')

Write 2D xi^s(s_perp, s_par) to file.

Parameters:

Name	Type	Description	Default
filename	str	Output path.	required
s_perp	ndarray	Unique s_perp values.	required
s_par	ndarray	Unique s_par values.	required
xi_s	(ndarray, shape(len(s_perp), len(s_par)))	2D correlation function.	required
header	str	Optional header string.	''

Source code in liulu/io/writers.py

def write_xi_2d(filename, s_perp, s_par, xi_s, header=""):
    """Write 2D xi^s(s_perp, s_par) to file.

    Parameters
    ----------
    filename : str
        Output path.
    s_perp : ndarray
        Unique s_perp values.
    s_par : ndarray
        Unique s_par values.
    xi_s : ndarray, shape (len(s_perp), len(s_par))
        2D correlation function.
    header : str
        Optional header string.
    """
    with open(filename, 'w') as f:
        f.write(f"# {header}\n" if header else "")
        f.write(f"# s_perp: {len(s_perp)} values from "
                f"{s_perp[0]:.4f} to {s_perp[-1]:.4f}\n")
        f.write(f"# s_par: {len(s_par)} values from "
                f"{s_par[0]:.4f} to {s_par[-1]:.4f}\n")
        f.write("# Columns: s_perp  s_par  xi_s\n")
        for i, sp in enumerate(s_perp):
            for j, sl in enumerate(s_par):
                f.write(f"{sp:12.6f} {sl:12.6f} {xi_s[i, j]:16.8e}\n")

liulu.io.writers.write_multipoles

write_multipoles(filename, s, multipoles_dict, header='')

Write multipoles to file.

Parameters:

Name	Type	Description	Default
filename	str	Output path.	required
s	ndarray	Separation values.	required
multipoles_dict	dict	{ell: xi_ell(s)} dictionary.	required
header	str	Optional header string.	''

Source code in liulu/io/writers.py

def write_multipoles(filename, s, multipoles_dict, header=""):
    """Write multipoles to file.

    Parameters
    ----------
    filename : str
        Output path.
    s : ndarray
        Separation values.
    multipoles_dict : dict
        {ell: xi_ell(s)} dictionary.
    header : str
        Optional header string.
    """
    ells = sorted(multipoles_dict.keys())
    col_header = "s  " + "  ".join(f"xi_{ell}" for ell in ells)

    data = np.column_stack([s] + [multipoles_dict[ell] for ell in ells])
    np.savetxt(filename, data,
               header=f"{header}\n{col_header}" if header else col_header,
               fmt='%16.8e')