dj_helper_fn.py

Helper functions for manipulating information from DataJoint fetch calls.

sanitize_unix_name(name)

Sanitize a string to be a valid unix name.

This function replaces any invalid characters with underscores and removes leading and trailing whitespace. It warns the user if the name has been changed.

Parameters:

Name	Type	Description	Default
name	str	Name to sanitize.	required

Returns:

Type	Description
str	Sanitized name.

Source code in src/spyglass/utils/dj_helper_fn.py

def sanitize_unix_name(name: str) -> str:
    """Sanitize a string to be a valid unix name.

    This function replaces any invalid characters with underscores and
    removes leading and trailing whitespace. It warns the user if the name
    has been changed.

    Parameters
    ----------
    name : str
        Name to sanitize.

    Returns
    -------
    str
        Sanitized name.
    """
    invalid_chars = r"[^a-zA-Z0-9_.-]+"
    sanitized_name = re.sub(invalid_chars, "_", name.strip().replace(" ", "_"))
    if sanitized_name != name:
        logger.warning(
            f"Name '{name}' contains invalid characters. "
            + f"Sanitized to '{sanitized_name}'."
        )
    return sanitized_name

ensure_names(table=None, force_list=False)

Ensure table is a string.

Parameters:

Name	Type	Description	Default
table	Union[str, Table, Iterable]	Table to ensure is a string, by default None. If passed as iterable, will ensure all elements are strings.	None
force_list	bool	Force the return to be a list, by default False, only used if input is iterable.	False

Returns:

Type	Description
Union[str, List[str], None]	Table as a string or list of strings.

Source code in src/spyglass/utils/dj_helper_fn.py

def ensure_names(
    table: Union[str, Table, Iterable] = None, force_list: bool = False
) -> Union[str, List[str], None]:
    """Ensure table is a string.

    Parameters
    ----------
    table : Union[str, Table, Iterable], optional
        Table to ensure is a string, by default None. If passed as iterable,
        will ensure all elements are strings.
    force_list : bool, optional
        Force the return to be a list, by default False, only used if input is
        iterable.

    Returns
    -------
    Union[str, List[str], None]
        Table as a string or list of strings.
    """
    # is iterable (list, set, set) but not a table/string
    is_collection = isinstance(table, Iterable) and not isinstance(
        table, (Table, TableMeta, str)
    )
    if force_list and not is_collection:
        return [ensure_names(table)]
    if table is None:
        return None
    if isinstance(table, str):
        return table
    if is_collection:
        return [ensure_names(t) for t in table]
    return getattr(table, "full_table_name", None)

declare_all_merge_tables()

Ensures all merge tables in the spyglass core package are declared.

• Prevents circular imports
• Prevents errors from table declaration within a transaction
• Run during nwb insertion

Source code in src/spyglass/utils/dj_helper_fn.py

def declare_all_merge_tables() -> Tuple[Type[dj.Table]]:
    """Ensures all merge tables in the spyglass core package are declared.

    - Prevents circular imports
    - Prevents errors from table declaration within a transaction
    - Run during nwb insertion
    """
    from spyglass.decoding.decoding_merge import DecodingOutput  # noqa: F401
    from spyglass.lfp.lfp_merge import LFPOutput  # noqa: F401
    from spyglass.position.position_merge import PositionOutput  # noqa: F401
    from spyglass.spikesorting.spikesorting_merge import (
        SpikeSortingOutput,
    )  # noqa: F401

    return DecodingOutput, LFPOutput, PositionOutput, SpikeSortingOutput

fuzzy_get(index, names, sources)

Given lists of items/names, return item at index or by substring.

Source code in src/spyglass/utils/dj_helper_fn.py

def fuzzy_get(index: Union[int, str], names: List[str], sources: List[str]):
    """Given lists of items/names, return item at index or by substring."""
    if isinstance(index, int):
        return sources[index]
    for i, part in enumerate(names):
        if index in part:
            return sources[i]
    return None

unique_dicts(list_of_dict)

Remove duplicate dictionaries from a list.

Source code in src/spyglass/utils/dj_helper_fn.py

def unique_dicts(list_of_dict):
    """Remove duplicate dictionaries from a list."""
    return [dict(t) for t in {tuple(d.items()) for d in list_of_dict}]

deprecated_factory(classes, old_module='')

Creates a list of classes and logs a warning when instantiated

Parameters:

Name	Type	Description	Default
classes	list	list of tuples containing old_class, new_class	required

Returns:

Type	Description
list	list of classes that will log a warning when instantiated

Source code in src/spyglass/utils/dj_helper_fn.py

def deprecated_factory(classes: list, old_module: str = "") -> list:
    """Creates a list of classes and logs a warning when instantiated

    Parameters
    ---------
    classes : list
        list of tuples containing old_class, new_class

    Returns
    ------
    list
        list of classes that will log a warning when instantiated
    """

    if not isinstance(classes, list):
        classes = [classes]

    ret = [
        _subclass_factory(old_name=c[0], new_class=c[1], old_module=old_module)
        for c in classes
    ]

    return ret[0] if len(ret) == 1 else ret

dj_replace(original_table, new_values, key_column, replace_column)

Given the output of a fetch() call from a schema and a 2D array made up of (key_value, replace_value) tuples, find each instance of key_value in the key_column of the original table and replace the specified replace_column with the associated replace_value. Key values must be unique.

Parameters:

Name	Type	Description	Default
original_table		Result of a datajoint .fetch() call on a schema query.	required
new_values	list	List of tuples, each containing (key_value, replace_value).	required
replace_column	str	The name of the column where to-be-replaced values are located.	required

Returns:

Type	Description
original_table	Structured array of new table entries that can be inserted back into the schema

Source code in src/spyglass/utils/dj_helper_fn.py

def dj_replace(original_table, new_values, key_column, replace_column):
    """Given the output of a fetch() call from a schema and a 2D array made up
    of (key_value, replace_value) tuples, find each instance of key_value in
    the key_column of the original table and replace the specified
    replace_column with the associated replace_value. Key values must be
    unique.

    Parameters
    ----------
    original_table
        Result of a datajoint .fetch() call on a schema query.
    new_values : list
        List of tuples, each containing (key_value, replace_value).
    replace_column : str
        The name of the column where to-be-replaced values are located.

    Returns
    -------
    original_table
        Structured array of new table entries that can be inserted back into the schema
    """

    # check to make sure the new_values are a list or array of tuples and fix if not
    if isinstance(new_values, tuple):
        tmp = list()
        tmp.append(new_values)
        new_values = tmp

    new_val_array = np.asarray(new_values)
    replace_ind = np.where(
        np.isin(original_table[key_column], new_val_array[:, 0])
    )
    original_table[replace_column][replace_ind] = new_val_array[:, 1]
    return original_table

get_all_tables_in_stack(stack)

Get all classes from a stack of tables.

Source code in src/spyglass/utils/dj_helper_fn.py

def get_all_tables_in_stack(stack):
    """Get all classes from a stack of tables."""
    classes = set()
    for frame_info in stack:
        locals_dict = frame_info.frame.f_locals
        for obj in locals_dict.values():
            if not isinstance(obj, UserTable):
                continue  # skip non-tables
            if (name := obj.full_table_name) in PERIPHERAL_TABLES:
                continue  # skip common_nwbfile tables
            classes.add(name)
    return classes

get_fetching_table_from_stack(stack)

Get all classes from a stack of tables.

Source code in src/spyglass/utils/dj_helper_fn.py

def get_fetching_table_from_stack(stack):
    """Get all classes from a stack of tables."""
    classes = get_all_tables_in_stack(stack)
    if len(classes) > 1:
        classes = None  # predict only one but not sure, so return None
    return next(iter(classes)) if classes else None

instance_table(table)

Instantiate a DataJoint table from its class, if uninstantiated.

Source code in src/spyglass/utils/dj_helper_fn.py

def instance_table(table: Union[str, Type[dj.Table]]) -> dj.Table:
    """Instantiate a DataJoint table from its class, if uninstantiated."""
    if isinstance(table, str):
        return dj.FreeTable(dj.conn(), table)
    if isinstance(table, type) and issubclass(table, dj.Table):
        return table()
    return table

fetch_nwb(*attrs, **kwargs)

Deprecated. Use (table & key).fetch_nwb() instead.

Migration guide: https://lorenfranklab.github.io/spyglass/latest/Features/Mixin/

Source code in src/spyglass/utils/dj_helper_fn.py

def fetch_nwb(*attrs, **kwargs):
    """Deprecated. Use ``(table & key).fetch_nwb()`` instead.

    Migration guide:
        https://lorenfranklab.github.io/spyglass/latest/Features/Mixin/
    """
    raise NotImplementedError(
        "fetch_nwb is deprecated. Use (table & key).fetch_nwb() instead."
    )

get_child_tables(table)

Get all child tables of a given table.

Source code in src/spyglass/utils/dj_helper_fn.py

def get_child_tables(table):
    """Get all child tables of a given table."""
    table = table() if inspect.isclass(table) else table
    return [
        dj.FreeTable(
            table.connection,
            (
                s
                if not s.isdigit()
                else next(iter(table.connection.dependencies.children(s)))
            ),
        )
        for s in table.children()
    ]

make_file_obj_id_unique(nwb_path)

Make the top-level object_id attribute of the file unique

Parameters:

Name	Type	Description	Default
nwb_path	str	path to the NWB file	required

Returns:

Type	Description
str	the new object_id

Source code in src/spyglass/utils/dj_helper_fn.py

def make_file_obj_id_unique(nwb_path: str):
    """Make the top-level object_id attribute of the file unique

    Parameters
    ----------
    nwb_path : str
        path to the NWB file

    Returns
    -------
    str
        the new object_id
    """
    from spyglass.common.common_lab import LabMember  # noqa: F401

    logger.info(f"Making unique object_id for {nwb_path}")
    LabMember().check_admin_privilege(
        error_message="Admin permissions required to edit existing analysis files"
    )
    new_id = str(uuid4())
    try:
        with h5py.File(nwb_path, "a") as f:
            f.attrs["object_id"] = new_id
    except (BlockingIOError, OSError):
        from spyglass.common.common_usage import ExportErrorLog

        ExportErrorLog().insert1(
            {
                "file": nwb_path,
                "source": "make_file_obj_id_unique",
            },
            skip_duplicates=True,
        )
        return
    location = "raw" if nwb_path.endswith("_.nwb") else "analysis"
    _resolve_external_table(
        nwb_path, nwb_path.split("/")[-1], location=location
    )
    return new_id

populate_pass_function(value)

Pass function for parallel populate.

Note: To avoid pickling errors, the table must be passed by class, NOT by instance. Note: This function must be defined in the global namespace.

Parameters:

Name	Type	Description	Default
value	(table, key, kwargs)	Class of table to populate, key to populate, and kwargs for populate	required

Source code in src/spyglass/utils/dj_helper_fn.py

def populate_pass_function(value):
    """Pass function for parallel populate.

    Note: To avoid pickling errors, the table must be passed by class,
        NOT by instance.
    Note: This function must be defined in the global namespace.

    Parameters
    ----------
    value : (table, key, kwargs)
        Class of table to populate, key to populate, and kwargs for populate
    """
    table, key, kwargs = value
    return table.populate(key, **kwargs)

NonDaemonPool

Bases: Pool

Non-daemonized pool for multiprocessing.

Used to create a pool of non-daemonized processes, which are required for parallel populate operations in DataJoint.

Source code in src/spyglass/utils/dj_helper_fn.py

class NonDaemonPool(multiprocessing.pool.Pool):
    """Non-daemonized pool for multiprocessing.

    Used to create a pool of non-daemonized processes, which are required for
    parallel populate operations in DataJoint.
    """

    # Explicitly set the start method to 'fork'
    # Allows the pool to be used in MacOS, where the default start method is 'spawn'
    multiprocessing.set_start_method("fork", force=True)

    def Process(self, *args, **kwds):
        """Return a non-daemonized process."""
        proc = super(NonDaemonPool, self).Process(*args, **kwds)

        class NonDaemonProcess(proc.__class__):
            """Monkey-patch process to ensure it is never daemonized"""

            @property
            def daemon(self):
                return False

            @daemon.setter
            def daemon(self, val):
                pass

        proc.__class__ = NonDaemonProcess
        return proc

Process(*args, **kwds)

Return a non-daemonized process.

Source code in src/spyglass/utils/dj_helper_fn.py

def Process(self, *args, **kwds):
    """Return a non-daemonized process."""
    proc = super(NonDaemonPool, self).Process(*args, **kwds)

    class NonDaemonProcess(proc.__class__):
        """Monkey-patch process to ensure it is never daemonized"""

        @property
        def daemon(self):
            return False

        @daemon.setter
        def daemon(self, val):
            pass

    proc.__class__ = NonDaemonProcess
    return proc

str_to_bool(value)

Return whether the provided string represents true. Otherwise false.

Source code in src/spyglass/utils/dj_helper_fn.py

def str_to_bool(value) -> bool:
    """Return whether the provided string represents true. Otherwise false."""
    # Due to distutils equivalent depreciation in 3.10
    # Adopted from github.com/PostHog/posthog/blob/master/posthog/utils.py
    if not value:
        return False
    return str(value).lower() in ("y", "yes", "t", "true", "1")

bytes_to_human_readable(size)

Convert a byte size to a human-readable format.

Source code in src/spyglass/utils/dj_helper_fn.py

def bytes_to_human_readable(size: int) -> str:
    """Convert a byte size to a human-readable format."""
    msg_template = "{size:.2f} {unit}"

    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if size < 1024:
            return msg_template.format(size=size, unit=unit)
        size /= 1024

    return msg_template.format(size=size, unit="PB")

accept_divergence(key, new_value, existing_value, test_mode=False, table_name=None)

Prompt to accept divergence in values between existing and new entries

Parameters:

Name	Type	Description	Default
key	str	Name of the column where the divergence is found	required
new_value	Any	New value to be inserted into the table	required
existing_value	Any	Existing value in the table that is different from the new value	required
test_mode	bool	If True, will not prompt and return False, by default False	False
table_name	str	Name of the table where the divergence is found, by default None	None

Source code in src/spyglass/utils/dj_helper_fn.py

def accept_divergence(
    key: str,
    new_value: Any,
    existing_value: Any,
    test_mode: bool = False,
    table_name: Optional[str] = None,
):
    """Prompt to accept divergence in values between existing and new entries

    Parameters
    ----------
    key : str
        Name of the column where the divergence is found
    new_value : Any
        New value to be inserted into the table
    existing_value : Any
        Existing value in the table that is different from the new value
    test_mode : bool, optional
        If True, will not prompt and return False, by default False
    table_name : str, optional
        Name of the table where the divergence is found, by default None
    """
    if test_mode:
        # If get here in test mode, is because want to test failure
        logger.debug(
            "\naccept_divergence called in testing, returning False w/o prompt"
        )
        return False
    tbl_msg = ""
    if table_name:  # optional message with table name
        tbl_msg = f" of '{table_name}'"
    response = dj.utils.user_choice(
        f"Existing entry differs in '{key}' column{tbl_msg}.\n"
        + "Accept the existing value of: \n"
        + f"'{existing_value}' \n"
        + "in place of the new value: \n"
        + f"'{new_value}' ?\n"
    )
    return str_to_bool(response)