Skip to content

plpipes.filesystem#

assign_section(target_section, relpath=None, section=None, **kwargs) #

Assign a target section in the configuration and return a constructed path.

Parameters:

Name Type Description Default
target_section str

The section to assign.

required
relpath str or None

Relative path to use if section does not exist.

None
section str or None

Alternative section to use.

None
**kwargs

Additional keyword arguments for the path function.

{}

Returns:

Name Type Description
Path

A Path object for the assigned section.

Source code in src\plpipes\filesystem.py
def assign_section(target_section, relpath=None, section=None, **kwargs):
    """
    Assign a target section in the configuration and return a constructed path.

    Args:
        target_section (str): The section to assign.
        relpath (str or None): Relative path to use if section does not exist.
        section (str or None): Alternative section to use.
        **kwargs: Additional keyword arguments for the path function.

    Returns:
        Path: A Path object for the assigned section.
    """
    if target_section not in cfg.cd("fs"):
        if relpath is None:
            relpath = target_section
        cfg["fs." + target_section] = str(_path(relpath, section))
    return path(section=target_section, **kwargs)

openfile(relpath, mode='r', section=None) #

Open a file and return the file object.

Parameters:

Name Type Description Default
relpath str

Relative path to the file.

required
mode str

Mode in which to open the file (default is 'r').

'r'
section str or None

Configuration section to use.

None

Returns:

Type Description

file object: The opened file object.

Source code in src\plpipes\filesystem.py
def openfile(relpath, mode="r", section=None):
    """
    Open a file and return the file object.

    Args:
        relpath (str): Relative path to the file.
        mode (str): Mode in which to open the file (default is 'r').
        section (str or None): Configuration section to use.

    Returns:
        file object: The opened file object.
    """
    return open(path(relpath, section), mode)

path(*args, mkparentdir=False, mkdir=False, **kwargs) #

Generate a Path object based on the provided arguments.

Parameters:

Name Type Description Default
*args

Positional arguments for the _path function.

()
mkparentdir bool

If True, create the parent directory of the path.

False
mkdir bool

If True, create the path directory.

False
**kwargs

Additional keyword arguments for the _path function.

{}

Returns:

Name Type Description
Path

A Path object representing the constructed path.

Source code in src\plpipes\filesystem.py
def path(*args, mkparentdir=False, mkdir=False, **kwargs):
    """
    Generate a Path object based on the provided arguments.

    Args:
        *args: Positional arguments for the _path function.
        mkparentdir (bool): If True, create the parent directory of the path.
        mkdir (bool): If True, create the path directory.
        **kwargs: Additional keyword arguments for the _path function.

    Returns:
        Path: A Path object representing the constructed path.
    """
    p = _path(*args, **kwargs)
    if mkdir:
        p.mkdir(exist_ok=True, parents=True)
    elif mkparentdir:
        p.parent.mkdir(exist_ok=True, parents=True)
    return p

read_csv(relpath, section=None, **kwargs) #

Read a CSV file and return a DataFrame.

Parameters:

Name Type Description Default
relpath str

Relative path to the CSV file.

required
section str or None

Configuration section to use.

None
**kwargs

Additional keyword arguments for pandas read_csv.

{}

Returns:

Name Type Description
DataFrame

DataFrame containing the CSV data.

Source code in src\plpipes\filesystem.py
def read_csv(relpath, section=None, **kwargs):
    """
    Read a CSV file and return a DataFrame.

    Args:
        relpath (str): Relative path to the CSV file.
        section (str or None): Configuration section to use.
        **kwargs: Additional keyword arguments for pandas read_csv.

    Returns:
        DataFrame: DataFrame containing the CSV data.
    """
    import pandas as pd
    return pd.read_csv(path(relpath, section), **kwargs)

read_excel(relpath, section=None, **kwargs) #

Read data from an Excel file and return a DataFrame.

Parameters:

Name Type Description Default
relpath str

Relative path to the Excel file.

required
section str or None

Configuration section to use.

None
**kwargs

Additional keyword arguments for pandas read_excel.

{}

Returns:

Name Type Description
DataFrame

DataFrame containing the Excel data.

Source code in src\plpipes\filesystem.py
def read_excel(relpath, section=None, **kwargs):
    """
    Read data from an Excel file and return a DataFrame.

    Args:
        relpath (str): Relative path to the Excel file.
        section (str or None): Configuration section to use.
        **kwargs: Additional keyword arguments for pandas read_excel.

    Returns:
        DataFrame: DataFrame containing the Excel data.
    """
    import pandas as pd
    return pd.read_excel(path(relpath, section), **kwargs)

read_json(relpath, section=None) #

Read data from a JSON file.

Parameters:

Name Type Description Default
relpath str

Relative path to the JSON file.

required
section str or None

Configuration section to use.

None

Returns:

Name Type Description
any

The deserialized data from the JSON file.

Source code in src\plpipes\filesystem.py
def read_json(relpath, section=None):
    """
    Read data from a JSON file.

    Args:
        relpath (str): Relative path to the JSON file.
        section (str or None): Configuration section to use.

    Returns:
        any: The deserialized data from the JSON file.
    """
    import json
    with openfile(relpath, section=section) as f:
        return json.load(f)

read_text(relpath, section=None, encoding='utf-8') #

Read text from a file.

Parameters:

Name Type Description Default
relpath str

Relative path to the text file.

required
section str or None

Configuration section to use.

None
encoding str

Encoding to use for reading the file (default is 'utf-8').

'utf-8'

Returns:

Name Type Description
str

The contents of the file as a string.

Source code in src\plpipes\filesystem.py
def read_text(relpath, section=None, encoding="utf-8"):
    """
    Read text from a file.

    Args:
        relpath (str): Relative path to the text file.
        section (str or None): Configuration section to use.
        encoding (str): Encoding to use for reading the file (default is 'utf-8').

    Returns:
        str: The contents of the file as a string.
    """
    target = path(relpath, section)
    with open(target, "r", encoding=encoding) as f:
        return f.read()

read_yaml(relpath, section=None) #

Read data from a YAML file.

Parameters:

Name Type Description Default
relpath str

Relative path to the YAML file.

required
section str or None

Configuration section to use.

None

Returns:

Name Type Description
any

The deserialized data from the YAML file.

Source code in src\plpipes\filesystem.py
def read_yaml(relpath, section=None):
    """
    Read data from a YAML file.

    Args:
        relpath (str): Relative path to the YAML file.
        section (str or None): Configuration section to use.

    Returns:
        any: The deserialized data from the YAML file.
    """
    import yaml
    with openfile(relpath, section=section) as f:
        return yaml.safe_load(f)

tempdir(parent=None) #

Create a temporary directory for file operations.

Parameters:

Name Type Description Default
parent str or None

Parent directory for the temporary directory.

None

Returns:

Name Type Description
TemporaryDirectory

Temporary directory context manager.

Source code in src\plpipes\filesystem.py
def tempdir(parent=None):
    """
    Create a temporary directory for file operations.

    Args:
        parent (str or None): Parent directory for the temporary directory.

    Returns:
        TemporaryDirectory: Temporary directory context manager.
    """
    if parent is None:
        parent = fs.path("tmp")
    import tempfile

    return tempfile.TemporaryDirectory(dir=parent)

write_csv(relpath, df, section=None, mkdir=True, **kwargs) #

Write a DataFrame to a CSV file.

Parameters:

Name Type Description Default
relpath str

Relative path for the CSV file.

required
df DataFrame

DataFrame to write.

required
section str or None

Configuration section to use.

None
mkdir bool

If True, create the directory if it does not exist.

True
**kwargs

Additional keyword arguments for pandas to_csv.

{}

Returns:

Type Description

None

Source code in src\plpipes\filesystem.py
def write_csv(relpath, df, section=None, mkdir=True, **kwargs):
    """
    Write a DataFrame to a CSV file.

    Args:
        relpath (str): Relative path for the CSV file.
        df (DataFrame): DataFrame to write.
        section (str or None): Configuration section to use.
        mkdir (bool): If True, create the directory if it does not exist.
        **kwargs: Additional keyword arguments for pandas to_csv.

    Returns:
        None
    """
    target = path(relpath, section)
    if mkdir:
        target.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(target, **kwargs)

write_excel(relpath, df, section=None, mkparentdir=True, autofilter=False, **kwargs) #

Write a DataFrame to an Excel file.

Parameters:

Name Type Description Default
relpath str

Relative path for the Excel file.

required
df DataFrame

DataFrame to write.

required
section str or None

Configuration section to use.

None
mkparentdir bool

If True, create the parent directory if it does not exist.

True
autofilter bool

If True, apply autofilter to the written Excel file.

False
**kwargs

Additional keyword arguments for pandas to_excel.

{}

Returns:

Name Type Description
Path

The path of the written Excel file.

Source code in src\plpipes\filesystem.py
def write_excel(relpath, df, section=None, mkparentdir=True, autofilter=False, **kwargs):
    """
    Write a DataFrame to an Excel file.

    Args:
        relpath (str): Relative path for the Excel file.
        df (DataFrame): DataFrame to write.
        section (str or None): Configuration section to use.
        mkparentdir (bool): If True, create the parent directory if it does not exist.
        autofilter (bool): If True, apply autofilter to the written Excel file.
        **kwargs: Additional keyword arguments for pandas to_excel.

    Returns:
        Path: The path of the written Excel file.
    """
    target = path(relpath, section=section, mkparentdir=mkparentdir)
    df.to_excel(target, index=False, **kwargs)

    if autofilter:
        import openpyxl
        wb = openpyxl.load_workbook(target)
        ws = wb.active
        ws.auto_filter.ref = ws.dimensions
        wb.save(target)

    return target

write_text(relpath, text, section=None, mkdir=True) #

Write text to a file.

Parameters:

Name Type Description Default
relpath str

Relative path for the text file.

required
text str

Text to write to the file.

required
section str or None

Configuration section to use.

None
mkdir bool

If True, create the directory if it does not exist.

True

Returns:

Name Type Description
int

Number of characters written to the file.

Source code in src\plpipes\filesystem.py
def write_text(relpath, text, section=None, mkdir=True):
    """
    Write text to a file.

    Args:
        relpath (str): Relative path for the text file.
        text (str): Text to write to the file.
        section (str or None): Configuration section to use.
        mkdir (bool): If True, create the directory if it does not exist.

    Returns:
        int: Number of characters written to the file.
    """
    target = path(relpath, section)
    if mkdir:
        target.parent.mkdir(parents=True, exist_ok=True)
    with open(target, "w") as f:
        return f.write(text)

write_yaml(relpath, data, section=None, mkdir=True, **kwargs) #

Write data to a YAML file.

Parameters:

Name Type Description Default
relpath str

Relative path for the YAML file.

required
data any

Data to serialize to YAML.

required
section str or None

Configuration section to use.

None
mkdir bool

If True, create the directory if it does not exist.

True
**kwargs

Additional keyword arguments for yaml.dump.

{}

Returns:

Type Description

None

Source code in src\plpipes\filesystem.py
def write_yaml(relpath, data, section=None, mkdir=True, **kwargs):
    """
    Write data to a YAML file.

    Args:
        relpath (str): Relative path for the YAML file.
        data (any): Data to serialize to YAML.
        section (str or None): Configuration section to use.
        mkdir (bool): If True, create the directory if it does not exist.
        **kwargs: Additional keyword arguments for yaml.dump.

    Returns:
        None
    """
    import yaml
    target = path(relpath, section)
    if mkdir:
        target.parent.mkdir(parents=True, exist_ok=True)
    with open(target, "w") as f:
        yaml.dump(data, f, **kwargs)