Create new data from old data in a child pipe

[p-zach]

Here's a code sample of how to use the sync function to perform calculations on existing data and put the resulting data in a child pipe.

This is a useful way to handle workflows where you're making changes to active plugins (more info here).

"""
Example of derivative sync plugins.
"""

required = ['pandas', 'random', 'datetime']

import meerschaum as mrsm
from meerschaum.utils.packages import lazy_import

# Lazy import so that it doesn't need to be loaded every time mrsm is called,
# only when it's needed
pd = lazy_import('pandas')

def register(pipe: mrsm.Pipe, **kw):
    return { 'columns': { 'datetime': 'timestamp' } }

def fetch(pipe: mrsm.Pipe, **kw):
    """Gets the data for the parent pipe.

    Args:
        pipe (mrsm.Pipe): The parent pipe.

    Returns:
        pandas.DataFrame: The dataframe of new data.
    """
    import random
    import datetime

    now = datetime.datetime.now()
    data = []

    # Generate 3 rows of random data
    for i in range(3):
        data.append({
            # Ensure distinct timestamps
            'timestamp': now + datetime.timedelta(seconds=i),
            'random1': random.randint(1, 100),
            'random2': random.randint(101, 200)
        })
        
    return pd.DataFrame(data)

def sync(pipe: mrsm.Pipe, **kw):
    """Syncs parent pipe, creates child pipe, derives 
    new data from the parent and syncs it with the child.

    Args:
        pipe (mrsm.Pipe): The parent pipe.

    Returns:
        mrsm.Pipe: The child pipe.
    """
    # Only continue if we're dealing with the parent pipe.
    if pipe.location_key is not None:
        return True, "Success"

    # Fetch data
    parent_data = fetch(pipe, **kw)
    # Sync parent pipe
    pipe.sync(parent_data, **kw)
    # Create child pipe, carrying over parent pipe's information
    # and setting the child pipe's location key to 'deriv_1'
    child_pipe = mrsm.Pipe(pipe.connector_keys, pipe.metric_key, 'deriv_1', columns = pipe.columns)
    # Get child data
    child_data = get_child_data(parent_data)
    # Add the fetched and additional data to the child pipe
    return child_pipe.sync(child_data, **kw)

def get_child_data(parent_data: 'pd.DataFrame'):
    """Derives new data from parent data.

    Args:
        parent_data (pandas.DataFrame): The new parent data.

    Returns:
        pandas.DataFrame: The child data, including the original data and the derived data.
    """
    # pandas.DataFrame.assign(...) is useful for deriving data.
    return parent_data.assign(
        # deriv_random1 and deriv_random2 will be the names of the new columns.
        deriv_random1=lambda row: row.random1 * 2,
        deriv_random2=lambda row: row.random2 + 0.5
    )

[bmeares]

@p-zach this is an excellent example of a development pattern for working with derivative pipes — nicely done! It's very readable and maintainable, which I sometimes struggle with when I think no one is watching👀

A couple small sharp edges: the pd.DataFrame type hint actually triggers the lazy_import, but you can avoid that by typing as a string (e.g. df: 'pd.DataFrame').

In fetch(), you build the Dataframe by calling pd.concat() in the loop. Not a problem for three rows, though for larger data sets, you can get better performance by appending DataFrames in a list and calling pd.concat() after the loop.

Finally, sync() returns a tuple of a bool and string (e.g. (True, 'Success'), which you can type with meerschaum.utils.typing.SuccessTuple).

I don't mean to nitpick! Your plugin is very polished and serves as a good framework for implementing future plugins in this development pattern. Using df.assign() with lambdas is an elegant solution that reads nicely, so I think I might start using that trick in my plugins.

Good work @p-zach 👏

[p-zach]

Thanks! Edited to fix those.

[bmeares]

About the SuccessTuple, I meant you only needed that if you were looking to add typing and for the docstring of sync() ― you can just return True, 'Success' in the if statement. But anyway, the edited plugin looks perfect! Appending a list of dictionaries in fetch() is a much more performant way of building a DataFrame than pd.concat().