🔍 Code Extractor

Extracts and flattens parameter names from a list of stream objects, with optional duplicate detection and parameter exclusion.

Purpose

This function is designed to collect parameter names from multiple stream objects into a single flat list. It's useful for introspecting stream configurations, validating parameter consistency across streams, and building parameter lists for downstream processing. The function can optionally detect and raise errors on duplicate parameter names across streams (except for Params stream types), and allows exclusion of specific parameter names like 'name' and '_memoize_key'.

Source Code

def stream_parameters(streams, no_duplicates=True, exclude=['name', '_memoize_key']):
    """
    Given a list of streams, return a flat list of parameter name,
    excluding those listed in the exclude list.

    If no_duplicates is enabled, a KeyError will be raised if there are
    parameter name clashes across the streams.
    """
    from ..streams import Params
    param_groups = {}
    for s in streams:
        if not s.contents and isinstance(s.hashkey, dict):
            param_groups[s] = list(s.hashkey)
        else:
            param_groups[s] = list(s.contents)

    if no_duplicates:
        seen, clashes = {}, []
        clash_streams = []
        for s in streams:
            if isinstance(s, Params):
                continue
            for c in param_groups[s]:
                if c in seen:
                    clashes.append(c)
                    if seen[c] not in clash_streams:
                        clash_streams.append(seen[c])
                    clash_streams.append(s)
                else:
                    seen[c] = s
        clashes = sorted(clashes)
        if clashes:
            clashing = ', '.join([repr(c) for c in clash_streams[:-1]])
            raise Exception('The supplied stream objects %s and %s '
                            'clash on the following parameters: %r'
                            % (clashing, clash_streams[-1], clashes))
    return [name for group in param_groups.values() for name in group
            if name not in exclude]

Parameters

Parameter Details

streams: A list or iterable of stream objects. Each stream should have either a 'contents' attribute (preferred) or a 'hashkey' dictionary attribute containing parameter names. The function extracts parameter names from these attributes.

no_duplicates: Boolean flag (default: True). When True, the function checks for parameter name clashes across different streams (excluding Params type streams). If duplicates are found, raises an Exception with details about which streams and parameters clash. When False, allows duplicate parameter names across streams.

exclude: List of parameter names to exclude from the final result (default: ['name', '_memoize_key']). These parameter names will be filtered out even if they exist in the stream objects.

Return Value

Returns a flat list of strings representing parameter names collected from all input streams. The list is created by iterating through all parameter groups from each stream and filtering out any names in the exclude list. If no_duplicates is True and clashes are detected, an Exception is raised instead of returning a value. The returned list may contain duplicates if no_duplicates is False.

Dependencies

• param

Required Imports

from ..streams import Params

Conditional/Optional Imports

These imports are only needed under specific conditions:

from ..streams import Params

Usage Example

# Assuming you have stream objects defined
from your_module.streams import Params

# Example stream objects (pseudo-code, actual implementation depends on your stream classes)
stream1 = SomeStreamClass()
stream1.contents = ['param1', 'param2', 'name']
stream1.hashkey = {'param1': 'value1', 'param2': 'value2'}

stream2 = SomeStreamClass()
stream2.contents = ['param3', 'param4']
stream2.hashkey = {'param3': 'value3', 'param4': 'value4'}

# Get all parameters excluding defaults
params = stream_parameters([stream1, stream2])
# Result: ['param1', 'param2', 'param3', 'param4']
# Note: 'name' is excluded by default

# Allow duplicates and include 'name'
params_with_dups = stream_parameters(
    [stream1, stream2],
    no_duplicates=False,
    exclude=['_memoize_key']
)

# This would raise an Exception due to duplicate 'param1'
stream3 = SomeStreamClass()
stream3.contents = ['param1', 'param5']
try:
    params = stream_parameters([stream1, stream3])
except Exception as e:
    print(f"Clash detected: {e}")

Best Practices

• Always ensure stream objects have either 'contents' or 'hashkey' attributes before passing to this function
• Use no_duplicates=True (default) when you need to ensure parameter uniqueness across streams to avoid configuration conflicts
• Params type streams are excluded from duplicate checking, allowing them to share parameter names with other streams
• The exclude list should contain parameter names that are metadata or internal identifiers rather than actual data parameters
• Handle the Exception that may be raised when no_duplicates=True and clashes are detected
• The function prioritizes 'contents' over 'hashkey' when both exist, so ensure 'contents' is properly populated if present

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function stream_name_mapping 63.0% similar

  Creates a bidirectional mapping dictionary between stream parameter names and their renamed versions, with optional parameter exclusion and reverse mapping capability.

  From: /tf/active/vicechatdev/patches/util.py

• function rename_stream_kwargs 59.6% similar

  Maps parameter names in a kwargs dictionary to their renamed equivalents for a given stream object, with optional reverse mapping from renamed back to original names.

  From: /tf/active/vicechatdev/patches/util.py

• function dimensionless_contents 56.0% similar

  Filters stream parameters to return only those that are not associated with any of the provided key dimensions.

  From: /tf/active/vicechatdev/patches/util.py

• function drop_streams 54.8% similar

  Filters out dimensioned stream parameters from key dimensions (kdims) and their corresponding keys, returning cleaned dimensions and keys.

  From: /tf/active/vicechatdev/patches/util.py

• function dimensioned_streams 50.5% similar

  Filters and returns streams from a DynamicMap that have parameters matching the DynamicMap's key dimensions.

  From: /tf/active/vicechatdev/patches/util.py