Skip to content

capturegraph.data.containers.utilities.zip #

zip - Broadcasting Zip for Lists#

Combine multiple Lists (or scalars) into a single List of Dicts for row-wise processing. Similar to zip(), but with NumPy-style broadcasting.

Example 
from pathlib import Path
from capturegraph.data import zip

# Build output paths from components
base = Path("./output")
names = sessions.date.map(lambda d: f"{d:%Y%m%d}.heic")

paths = zip(dir=base, name=names).map(lambda r: r.dir / r.name)
# → List([Path("./output/20260104.heic"), ...])

# Copy files using zip
import shutil
zip(src=sessions.photo, dst=paths).map(
    lambda r: shutil.copy2(r.src, r.dst)
)
Broadcasting Rules
  • Same length: All Lists must have the same length
  • Single-element: Lists with one element are repeated to match
  • Scalars: Non-list values are repeated for every row
zip(a=[1,2,3], b=[4,5,6])  # → [{'a':1,'b':4}, {'a':2,'b':5}, {'a':3,'b':6}]
zip(a=[1,2,3], b=99)       # → [{'a':1,'b':99}, {'a':2,'b':99}, {'a':3,'b':99}]
zip(a=[1], b=[4,5,6])      # → [{'a':1,'b':4}, {'a':1,'b':5}, {'a':1,'b':6}]

zip(*args, **kwargs) #

Combine multiple Lists with NumPy-style broadcasting.

Creates a List of Dicts, where each Dict represents one row with named fields from the keyword arguments.

Parameters:

Name Type Description Default
**kwargs T | List[T]

Named Lists or scalars. - Lists are zipped element-wise. - Scalars are repeated for every row. - Single-element lists are broadcast to match the longest.

 {}

Returns:

Type Description
List[Dict[Any]]

List of Dicts with attribute access to each field.

Raises:

Type Description
ValueError

If List lengths are incompatible (different and not 1).
Example 
# Build file paths
paths = zip(
    dir=Path("./output"),
    name=sessions.date.map(lambda d: f"{d:%Y%m%d}.heic")
).map(lambda r: r.dir / r.name)

# Copy files
import shutil
zip(src=sessions.photo, dst=paths).map(
    lambda r: shutil.copy2(r.src, r.dst)
)
Source code in capturegraph-lib/capturegraph/data/containers/utilities/zip.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def zip[T](*args: T | List[T], **kwargs: T | List[T]) -> List[Dict[Any]]:
    """Combine multiple Lists with NumPy-style broadcasting.

    Creates a List of Dicts, where each Dict represents one row
    with named fields from the keyword arguments.

    Args:
        **kwargs: Named Lists or scalars.
            - Lists are zipped element-wise.
            - Scalars are repeated for every row.
            - Single-element lists are broadcast to match the longest.

    Returns:
        List of Dicts with attribute access to each field.

    Raises:
        ValueError: If List lengths are incompatible (different and not 1).

    Example:
        ```python
        # Build file paths
        paths = zip(
            dir=Path("./output"),
            name=sessions.date.map(lambda d: f"{d:%Y%m%d}.heic")
        ).map(lambda r: r.dir / r.name)

        # Copy files
        import shutil
        zip(src=sessions.photo, dst=paths).map(
            lambda r: shutil.copy2(r.src, r.dst)
        )
        ```
    """
    if not args and not kwargs:
        raise ValueError("zip requires at least one argument")

    args = {f"{i}": v for i, v in enumerate(args)} | kwargs

    values = list(args.values())
    keys = list(args.keys())

    length = _broadcast_length(*values)
    broadcasted = [_broadcast_value(v, length) for v in values]

    rows: list[Dict[Any]] = []
    for i in range(length):
        row_data = {key: broadcasted[j][i] for j, key in enumerate(keys)}
        rows.append(Dict(row_data))

    return List(rows)