populate_all_common.py

log_insert_error(table, err, error_constants=None)

Log a given error to the InsertError table.

Parameters:

Name	Type	Description	Default
table	str	The table name where the error occurred.	required
err	Exception	The exception that was raised.	required
error_constants	dict	Dictionary with keys for dj_user, connection_id, and nwb_file_name. Defaults to checking dj.conn and using "Unknown" for nwb_file_name.	None

Source code in src/spyglass/common/populate_all_common.py

def log_insert_error(
    table: str, err: Exception, error_constants: dict = None
) -> None:
    """Log a given error to the InsertError table.

    Parameters
    ----------
    table : str
        The table name where the error occurred.
    err : Exception
        The exception that was raised.
    error_constants : dict, optional
        Dictionary with keys for dj_user, connection_id, and nwb_file_name.
        Defaults to checking dj.conn and using "Unknown" for nwb_file_name.
    """
    if error_constants is None:
        error_constants = dict(
            dj_user=dj.config["database.user"],
            connection_id=dj.conn().connection_id,
            nwb_file_name="Unknown",
        )
    InsertError.insert1(
        dict(
            **error_constants,
            table=table.__name__,
            error_type=type(err).__name__,
            error_message=str(err)[:255],  # limit to 255 chars
            error_raw=str(err),
        )
    )

single_transaction_make(tables, nwb_file_name, raise_err=False, error_constants=None, config=None)

Ingest each table from the NWB file, inside one transaction.

Every table here is a SpyglassIngestion table, so each parses the file once via insert_from_nwbfile rather than running make per key_source key. Failures are logged per table unless raise_err is set.

Source code in src/spyglass/common/populate_all_common.py

def single_transaction_make(
    tables: List[dj.Table],
    nwb_file_name: str,
    raise_err: bool = False,
    error_constants: dict = None,
    config: dict = None,
):
    """Ingest each table from the NWB file, inside one transaction.

    Every table here is a SpyglassIngestion table, so each parses the file
    once via `insert_from_nwbfile` rather than running `make` per key_source
    key. Failures are logged per table unless `raise_err` is set.
    """

    # Entries may also be declared in a `_spyglass_config.yaml` beside the NWB
    # file. Both configs share the {TableName: [rows]} shape that
    # `generate_entries_from_config` indexes by name, so each table is handed
    # the whole merged mapping -- a per-table lookup would yield a row list.
    # The file's own config wins: `entries.yaml` holds lab-wide defaults,
    # while the sidecar describes this session. Before this PR the sidecar was
    # the only config any table that read one consulted.
    # `or dict()`: yaml.safe_load returns None for an empty file, and the
    # config argument is optional.
    file_config = (
        get_config(
            Nwbfile.get_abs_path(nwb_file_name),
            calling_table="populate_all_common",
        )
        or dict()
    )
    merged_config = {**(config or dict()), **file_config}

    with Nwbfile._safe_context():
        for table in tables:
            try:
                table().insert_from_nwbfile(nwb_file_name, config=merged_config)
            except Exception as err:
                if raise_err:
                    raise err
                log_insert_error(
                    table=table, err=err, error_constants=error_constants
                )

populate_all_common(nwb_file_name, rollback_on_fail=False, raise_err=False)

Insert all common tables for a given NWB file.

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The name of the NWB file to populate.	required
rollback_on_fail	bool	If True, will delete the Session entry if any errors occur. Defaults to False.	False
raise_err	bool	If True, will raise any errors that occur during population. Defaults to False. This will prevent any rollback from occurring.	False

Returns:

Type	Description
List	A list of keys for InsertError entries if any errors occurred.

Notes

InsertError rows logged by an earlier attempt at the same file, under the same user and connection, are cleared before population starts, so the returned list only ever describes the current attempt.

Source code in src/spyglass/common/populate_all_common.py

def populate_all_common(
    nwb_file_name, rollback_on_fail=False, raise_err=False
) -> Union[List, None]:
    """Insert all common tables for a given NWB file.

    Parameters
    ----------
    nwb_file_name : str
        The name of the NWB file to populate.
    rollback_on_fail : bool, optional
        If True, will delete the Session entry if any errors occur.
        Defaults to False.
    raise_err : bool, optional
        If True, will raise any errors that occur during population.
        Defaults to False. This will prevent any rollback from occurring.

    Returns
    -------
    List
        A list of keys for InsertError entries if any errors occurred.

    Notes
    -----
    InsertError rows logged by an earlier attempt at the same file, under the
    same user and connection, are cleared before population starts, so the
    returned list only ever describes the current attempt.
    """
    from spyglass.lfp.lfp_imported import ImportedLFP
    from spyglass.position.v1.imported_pose import ImportedPose
    from spyglass.spikesorting.imported import ImportedSpikeSorting

    _ = declare_all_merge_tables()

    error_constants = dict(
        dj_user=dj.config["database.user"],
        connection_id=dj.conn().connection_id,
        nwb_file_name=nwb_file_name,
    )

    # Drop errors logged by an earlier attempt at this same file, user, and
    # connection. Without this, the check below reports stale failures and can
    # roll back an otherwise clean ingestion. See issue #1497. InsertError has
    # no dependent tables, so delete_quick is safe here.
    (InsertError & error_constants).delete_quick()

    table_lists: List[List[dj.Table]] = [
        # Tables that can be inserted in a single transaction
        [
            Institution,  # Parent node
            Lab,  # Parent node
            LabMember,  # Parent node
            LabTeam,  # Parent node
            Subject,  # Parent node
            CameraDevice,  # Parent node
            ProbeType,  # Parent node
            DataAcquisitionDeviceAmplifier,  # Parent node
            DataAcquisitionDeviceSystem,  # Parent node
            DataAcquisitionDevice,  # Depends on DataAcq*Amp, DataAcq*Sys
            OpticalFiberDevice,  # Parent node
            Virus,  # Parent node
        ],
        [
            Probe,  # Depends on ProbeType, DataAcquisitionDevice
            Probe.Shank,  # Depends on Probe
            Probe.Electrode,  # Depends on Probe
            Session,  # Depends on Subject, Institution, Lab
            Session.Experimenter,  # Depends on Session
            Session.DataAcquisitionDevice,  # Depends on Sess, DataAcq*Device
            ElectrodeGroup,  # Depends on Session
            Raw,  # Depends on Session
            SampleCount,  # Depends on Session
            DIOEvents,  # Depends on Session
            ImportedSpikeSorting,  # Depends on Session
            SensorData,  # Depends on Session
            IntervalList,  # Depends on Session
            TaskEpoch,  # Depends on Session, Task, CamearaDevice, IntervalList
            # NwbfileKachery, # Not used by default
        ],
        [  # Tables that depend on above transaction
            Electrode,  # Depends on ElectrodeGroup
            PositionSource,  # Depends on Session. Also fills RawPosition
            RawCompassDirection,  # Depends on Session
            VideoFile,  # Depends on TaskEpoch
            StateScriptFile,  # Depends on TaskEpoch
            ImportedPose,  # Depends on Session
            ImportedLFP,  # Depends on ElectrodeGroup
            VirusInjection,  # Depends on Session
            OpticalFiberImplant,  # Depends on Session and OpticalFiberDevice
            OptogeneticProtocol,  # Depends on Session and TaskEpoch
        ],
    ]

    config = dict()
    entries_path = Path(base_dir) / "entries.yaml"
    if entries_path.exists():
        with open(f"{base_dir}/entries.yaml", "r") as stream:
            config = yaml.safe_load(stream)

    for tables in table_lists:
        single_transaction_make(
            tables=tables,
            nwb_file_name=nwb_file_name,
            raise_err=raise_err,
            error_constants=error_constants,
            config=config,
        )

    err_query = InsertError & error_constants
    nwbfile_query = Nwbfile & {"nwb_file_name": nwb_file_name}

    if err_query and nwbfile_query and rollback_on_fail:
        logger.error(f"Rolling back population for {nwb_file_name}...")
        # Should this be safemode=False to prevent confirmation prompt?
        nwbfile_query.super_delete(warn=False)

    if err_query:
        err_tables = err_query.fetch("table")
        logger.error(
            f"Errors occurred during population for {nwb_file_name}:\n\t"
            + f"Failed tables {err_tables}\n\t"
            + "See common_usage.InsertError for more details"
        )
        return err_query.fetch("KEY")