Core

connect(streams, uri=None)

Connect hetegeneous streams into a single stream.

Unlike stream.join, this function can connect streams with unique params, such as different renderers.

Parameters:

Name	Type	Description	Default
streams	list[AnyStream | GifStream | Mp4Stream | HtmlStream]	The streams to connect.	required
uri	str | Path | BytesIO | None	The destination to write the connected streams to. If None, the connected streams are returned.	None

Returns:

Type	Description
ConnectedStreams | Path	The connected streams if uri is None, otherwise the uri.

Source code in streamjoy/core.py

def connect(
    streams: list[AnyStream | GifStream | Mp4Stream | HtmlStream],
    uri: str | Path | BytesIO | None = None,
) -> ConnectedStreams | Path:
    """
    Connect hetegeneous streams into a single stream.

    Unlike `stream.join`, this function can connect streams
    with unique params, such as different renderers.

    Args:
        streams: The streams to connect.
        uri: The destination to write the connected streams to.
            If None, the connected streams are returned.

    Returns:
        The connected streams if uri is None, otherwise the uri.
    """
    stream = ConnectedStreams(streams=streams)
    if uri:
        return stream.write(uri=uri)
    return stream

stream(resources, uri=None, renderer=None, renderer_iterables=None, renderer_kwargs=None, extension=None, **kwargs)

Create a stream from the given resources.

Parameters:

Name	Type	Description	Default
resources	Any	The resources to create a stream from.	required
uri	str | Path | BytesIO | None	The destination to write the stream to. If None, the stream is returned.	None
renderer	Callable | None	The renderer to use. If None, the default renderer is used.	None
renderer_iterables	list | None	Additional positional arguments to map over the renderer.	None
renderer_kwargs	dict | None	Additional keyword arguments to pass to the renderer.	None
extension	Literal['.mp4', '.gif'] | None	The extension to use; useful if uri is a file-like object.	None
**kwargs	dict[str, Any]	Additional keyword arguments to pass.	{}

Returns:

Type	Description
AnyStream | GifStream | Mp4Stream | HtmlStream | Path	The stream if uri is None, otherwise the uri.

Source code in streamjoy/core.py

def stream(
    resources: Any,
    uri: str | Path | BytesIO | None = None,
    renderer: Callable | None = None,
    renderer_iterables: list | None = None,
    renderer_kwargs: dict | None = None,
    extension: Literal[".mp4", ".gif"] | None = None,
    **kwargs: dict[str, Any],
) -> AnyStream | GifStream | Mp4Stream | HtmlStream | Path:
    """
    Create a stream from the given resources.

    Args:
        resources: The resources to create a stream from.
        uri: The destination to write the stream to. If None, the stream is returned.
        renderer: The renderer to use. If None, the default renderer is used.
        renderer_iterables: Additional positional arguments to map over the renderer.
        renderer_kwargs: Additional keyword arguments to pass to the renderer.
        extension: The extension to use; useful if uri is a file-like object.
        **kwargs: Additional keyword arguments to pass.

    Returns:
        The stream if uri is None, otherwise the uri.
    """
    if isinstance(uri, str):
        uri = Path(uri)

    extension = extension or (uri and uri.suffix)
    if isinstance(extension, str) and extension not in extension_handlers:
        raise ValueError(f"Unsupported extension: {extension}")

    stream_cls = getattr(streams, extension_handlers.get(extension), AnyStream)
    serialized = serialize_appropriately(
        stream_cls,
        resources,
        renderer,
        renderer_iterables,
        renderer_kwargs or {},
        **kwargs,
    )
    stream = stream_cls(**serialized.param.values(), **serialized.kwargs)

    if uri:
        return stream.write(uri=uri, extension=extension)
    return stream