settings.py

SpyglassConfig

Gets Spyglass dirs from dj.config or environment variables.

Uses SpyglassConfig.relative_dirs to (a) gather user settings from dj.config or os environment variables or defaults relative to base, in that order (b) set environment variables, and (c) make dirs that don't exist. NOTE: when passed a base_dir, it will ignore env vars to facilitate testing.

Source code in src/spyglass/settings.py

class SpyglassConfig:
    """Gets Spyglass dirs from dj.config or environment variables.

    Uses SpyglassConfig.relative_dirs to (a) gather user
    settings from dj.config or os environment variables or defaults relative to
    base, in that order (b) set environment variables, and (c) make dirs that
    don't exist. NOTE: when passed a base_dir, it will ignore env vars to
    facilitate testing.
    """

    @staticmethod
    def _load_directory_schema():
        """Load directory schema from JSON file in package directory.

        Returns
        -------
        dict
            Directory schema with prefixes (spyglass, kachery, dlc, moseq)

        Raises
        ------
        FileNotFoundError
            If directory_schema.json is not found in spyglass package
        ValueError
            If schema is invalid or missing required keys

        Notes
        -----
        This method reads from directory_schema.json in the spyglass package,
        which is the single source of truth for Spyglass directory structure.
        """
        schema_path = Path(__file__).parent / "directory_schema.json"

        if not schema_path.exists():
            raise FileNotFoundError(
                f"Config schema file not found at {schema_path}. "
                "This file is required for Spyglass to function. "
                "Please ensure you have a complete Spyglass installation."
            )

        with open(schema_path) as f:
            schema = json.load(f)

        if not isinstance(schema, dict):
            raise ValueError(f"Schema should be a dict, got {type(schema)}")

        if "directory_schema" not in schema:
            raise ValueError("Schema missing 'directory_schema' key")

        # Note: _schema_version field is informational only, not enforced
        return schema["directory_schema"]

    def __init__(self, base_dir: str = None, **kwargs) -> None:
        """
        Initializes a new instance of the class.

        Parameters
        ----------
        base_dir (str)
            The base directory.

        Attributes
        ----------
        supplied_base_dir (str)
            The base directory passed to the class.
        config_defaults (dict)
            Default settings for the config.
        relative_dirs (dict)
            Relative dirs for each prefix (spyglass, kachery, dlc). Relative
            to respective base_dir. Created on init.
        dj_defaults (dict)
            Default settings for datajoint.
        env_defaults (dict)
            Default settings for environment variables.
        _config (dict)
            Cached config settings.
        _debug_mode (bool)
            True if debug_mode is set. Supports skipping known bugs in test env.
        _test_mode (bool or object)
            The bound test mode, or ``_UNSET`` before the first deliberate or
            successful load. The public ``test_mode`` property always returns
            a bool.
        """
        self.supplied_base_dir = base_dir
        self._config = dict()
        self.config_defaults = dict(prepopulate=True)
        # Constructor values participate in first-load precedence. test_mode
        # becomes instance identity once an explicit load starts or an ambient
        # load succeeds; debug_mode remains an ordinary reloadable setting.
        self._debug_mode_arg = kwargs.get("debug_mode", _UNSET)
        self._initial_test_mode = kwargs.get("test_mode", _UNSET)
        self._debug_mode = (
            False
            if self._debug_mode_arg is _UNSET
            else str_to_bool(self._debug_mode_arg)
        )
        self._test_mode = _UNSET
        self._dlc_base = None
        # Initialized here, not only in load_config's COMMIT phase: a load
        # that fails or returns early (e.g. no base under an ambient test
        # mode) still leaves `_dj_custom`/`_generate_dj_config` able to read
        # it, matching `_dlc_base`.
        self._moseq_base = None
        self.load_failed = False
        # A mode-change request invalidates a loaded instance permanently. Keep
        # only the message (not an exception/traceback); recovery uses a new
        # SpyglassConfig object with an unambiguous lifecycle.
        self._mode_error: str | None = None

        # Load directory schema from JSON file (single source of truth)
        # {PREFIX}_{KEY}_DIR, default dir relative to base_dir
        # NOTE: Adding new dir requires edit to HHMI hub AND directory_schema.json
        self.relative_dirs = self._load_directory_schema()
        self.dj_defaults = {
            "database.host": kwargs.get("database_host", "lmf-db.cin.ucsf.edu"),
            "database.user": kwargs.get("database_user"),
            "database.port": kwargs.get("database_port", 3306),
            "database.use_tls": kwargs.get("database_use_tls", True),
            "filepath_checksum_size_limit": 1 * 1024**3,
            "enable_python_native_blobs": True,
        }
        self.env_defaults = {
            "FIGURL_CHANNEL": "franklab2",
            "DJ_SUPPORT_FILEPATH_MANAGEMENT": "TRUE",
            "KACHERY_CLOUD_EPHEMERAL": "TRUE",
            "HDF5_USE_FILE_LOCKING": "FALSE",
        }

    def _resolve_test_mode(self, call_value, dj_custom) -> tuple[bool, bool]:
        """Return this instance's test mode and whether it is bound.

        First-load precedence is call argument, constructor argument,
        ``dj.config['custom']``, then ``False``. A call or constructor value
        binds before path validation, so a failed explicit test-mode load
        cannot later retry implicitly in production mode. A successful ambient
        load is bound during commit. Once bound, the mode is immutable.
        """
        if self._mode_error is not None:
            raise ValueError(self._mode_error)

        if self._test_mode is not _UNSET:
            bound_mode = self._test_mode
            if call_value is not _UNSET:
                requested_mode = str_to_bool(call_value)
                if requested_mode != bound_mode:
                    message = (
                        "SpyglassConfig test_mode is already bound to "
                        f"{bound_mode} and cannot change to {requested_mode}; "
                        "create a new SpyglassConfig instance."
                    )
                    # Do not leave the old paths usable after a caller has
                    # explicitly requested a different safety mode and caught
                    # the rejection.
                    self._config = {}
                    self.load_failed = True
                    self._mode_error = message
                    raise ValueError(message)
            return bound_mode, True

        if call_value is not _UNSET:
            test_mode = str_to_bool(call_value)
        elif self._initial_test_mode is not _UNSET:
            test_mode = str_to_bool(self._initial_test_mode)
        else:
            # Ambient state stays unbound until a configuration commits. This
            # lets import-time startup remain graceful when no base exists.
            return str_to_bool(dj_custom.get("test_mode", False)), False

        # Bind deliberate mode before validation. In particular, a failed
        # explicit test load must not let a later property access consult a
        # production SPYGLASS_BASE_DIR.
        self._test_mode = test_mode
        return test_mode, True

    def load_config(
        self,
        base_dir=None,
        force_reload=False,
        on_startup: bool = False,
        **kwargs,
    ) -> dict | None:
        """
        Loads the configuration settings for the object.

        Order of precedence, where X is base, raw, analysis, etc.:
        1. SpyglassConfig(base_dir="string") for base dir only
        2. dj.config['custom']['spyglass_dirs']['X']
        3. dj.config['custom']['kachery_dirs']['X']
        4. os.environ['{SPYGLASS/KACHERY}_{X}_DIR']
        5. resolved_base_dir/X for non-base dirs

        When test_mode=True, environment variables are not consulted for any
        directory path, and the resolved base_dir must contain a 'tests' path
        component.

        ``test_mode`` binds to the instance on the first deliberate or
        successful load and is then immutable. Passing a ``test_mode`` that
        differs from the bound value (even without ``force_reload``) does not
        transition the instance -- it invalidates it and raises. To switch
        modes, construct a new ``SpyglassConfig``.

        Parameters
        ----------
        base_dir: str
            Optional. Default None. The base directory. If not provided, will
            use the env variable or existing config.
        force_reload: bool
            Optional. Default False. Default skip load if already completed.

        Raises
        ------
        ValueError
            When a caller attempts to change the mode of a bound instance; or,
            under test_mode, when a deliberate load cannot resolve a base_dir,
            the resolved base_dir does not contain a 'tests' path component, or
            any resolved directory -- including one reached through a symlink
            -- falls outside that base_dir. A fresh ambient (dj.config-only)
            test-mode load with no base returns gracefully instead.

        Returns
        -------
        dict
            list of relative_dirs and other settings (e.g., prepopulate).
        """
        # Fast path for the common cached read (every directory property
        # routes here with no kwargs). A mode-change request (explicit
        # test_mode=) still falls through to _resolve_test_mode's binding /
        # rejection, and a mode-wedged instance has _config == {} (falsy) so it
        # also falls through and re-raises.
        if (
            not force_reload
            and self._config
            and kwargs.get("test_mode", _UNSET) is _UNSET
        ):
            return self._config

        dj_custom = dj.config.get("custom", {})
        dj_spyglass = dj_custom.get("spyglass_dirs", {})
        dj_kachery = dj_custom.get("kachery_dirs", {})
        dj_dlc = dj_custom.get("dlc_dirs", {})
        dj_moseq = dj_custom.get("moseq_dirs", {})

        test_mode, test_mode_is_bound = self._resolve_test_mode(
            kwargs.get("test_mode", _UNSET), dj_custom
        )
        if not force_reload and self._config:
            return self._config

        def _resolve_debug_mode() -> bool:
            """Resolve call > constructor > DataJoint > default precedence."""
            call_value = kwargs.get("debug_mode", _UNSET)
            if call_value is not _UNSET:
                return str_to_bool(call_value)
            if self._debug_mode_arg is not _UNSET:
                return str_to_bool(self._debug_mode_arg)
            return str_to_bool(dj_custom.get("debug_mode", False))

        debug_mode = _resolve_debug_mode()

        # Until a deliberate test-mode load commits, keep the object visibly
        # failed. A successful commit below resets this flag. Same-mode reloads
        # of an existing valid test config remain transactional.
        if test_mode and test_mode_is_bound and not self._config:
            self.load_failed = True

        resolved_base = (
            base_dir
            or self.supplied_base_dir
            or dj_spyglass.get("base")
            # Gated by test_mode like every other directory env var below:
            # SPYGLASS_BASE_DIR is the exact production path this sandbox
            # exists to keep destructive tests off, so test_mode must not
            # inherit it. Explicit base_dir/config still resolve.
            or (None if test_mode else os.environ.get("SPYGLASS_BASE_DIR"))
        )

        # Log when supplied base_dir causes environment variable overrides to be ignored
        if self.supplied_base_dir:
            logger.info(
                "Using supplied base_dir - ignoring SPYGLASS_* environment variable overrides"
            )

        # ---------------------------- RESOLVE ----------------------------
        # Compute every path as a plain value. Nothing is created and no
        # external/global state is mutated until validation passes.
        if not resolved_base:
            self.load_failed = True
            if test_mode and test_mode_is_bound:
                raise ValueError(
                    "Refusing to load Spyglass in test_mode without an "
                    "explicit base_dir or "
                    "dj.config['custom']['spyglass_dirs']['base']; "
                    "SPYGLASS_BASE_DIR is ignored in test_mode."
                )
            if not on_startup:  # Only warn if not on startup
                logger.error(
                    "Could not find SPYGLASS_BASE_DIR"
                    + "\n\tCheck dj.config['custom']['spyglass_dirs']['base']"
                    + "\n\tand os.environ['SPYGLASS_BASE_DIR']"
                )
            return

        base_path = Path(resolved_base).expanduser().resolve()
        resolved_base = str(base_path)

        def env_or_none(var: str) -> str | None:
            """Read an env var, ignored in test_mode to keep the sandbox."""
            return None if test_mode else os.environ.get(var)

        dlc_project = env_or_none("DLC_PROJECT_PATH")
        dlc_base = (
            dj_dlc.get("base")
            or env_or_none("DLC_BASE_DIR")
            or (dlc_project.split("projects")[0] if dlc_project else None)
            or str(Path(resolved_base) / "deeplabcut")
        )
        moseq_base = (
            dj_moseq.get("base")
            or env_or_none("MOSEQ_BASE_DIR")
            or str(Path(resolved_base) / "moseq")
        )

        config_dirs = {"SPYGLASS_BASE_DIR": str(resolved_base)}
        source_config_lookup = {
            "dlc": dj_dlc,
            "moseq": dj_moseq,
            "kachery": dj_kachery,
        }
        base_lookup = {"dlc": dlc_base, "moseq": moseq_base}
        for prefix, dirs in self.relative_dirs.items():
            this_base = base_lookup.get(prefix, resolved_base)
            for dir, dir_str in dirs.items():
                dir_env_fmt = self.dir_to_var(dir=dir, dir_type=prefix)

                env_loc = (  # Ignore env vars if base was passed or test_mode
                    os.environ.get(dir_env_fmt)
                    if not self.supplied_base_dir and not test_mode
                    else None
                )
                source_config = source_config_lookup.get(prefix, dj_spyglass)
                dir_location = (
                    source_config.get(dir)
                    or env_loc
                    or str(Path(this_base) / dir_str)
                ).replace('"', "")

                config_dirs.update({dir_env_fmt: str(dir_location)})

        kachery_zone_dict = {
            "KACHERY_ZONE": (
                env_or_none("KACHERY_ZONE")
                or dj.config.get("custom", {}).get("kachery_zone")
                or "franklab.default"
            )
        }

        # ---------------------------- VALIDATE ---------------------------
        # Both checks apply ONLY under test_mode. Production configuration
        # is unchanged: an analysis dir anywhere, including behind a
        # symlink, stays legal.
        if test_mode:
            validation_error = None
            if "tests" not in base_path.parts:
                validation_error = (
                    f"Refusing to load Spyglass in test_mode with base_dir "
                    f"{resolved_base!r}: path does not contain a 'tests' "
                    "component. Run pytest with --base-dir pointing inside a "
                    "tests/ directory (default: ./tests/_data/) to keep "
                    "destructive operations off shared/production storage."
                )
            else:
                # Path.resolve() is non-strict: a dir that does not exist yet
                # resolves to its would-be path, while an EXISTING symlink
                # resolves through to its target. That is what catches an
                # analysis dir symlinked at production storage.
                checked = dict(config_dirs)
                checked["DLC_BASE_DIR"] = dlc_base
                checked["MOSEQ_BASE_DIR"] = moseq_base
                for var, loc in checked.items():
                    loc_path = Path(loc).expanduser().resolve()
                    if not loc_path.is_relative_to(base_path):
                        validation_error = (
                            f"Refusing to load Spyglass in test_mode: {var} "
                            f"resolves to {str(loc_path)!r}, outside the test "
                            f"base {resolved_base!r}. Destructive tests must "
                            "stay within the test base directory; check "
                            "dj.config['custom'] and any directory symlinks."
                        )
                        break

            if validation_error is not None:
                # A deliberate (bound) test-mode load fails loud. An ambient /
                # implicit load must not crash unrelated code -- matching the
                # no-base handling above -- but must also NOT commit a
                # test-mode config whose paths escape the sandbox. So mark the
                # load failed and return without committing, leaving the mode
                # unbound.
                self.load_failed = True
                if test_mode_is_bound:
                    raise ValueError(validation_error)
                if not on_startup:  # Only warn if not on startup
                    logger.error(validation_error)
                return

        # ----------------------------- COMMIT ----------------------------
        if self._test_mode is _UNSET:
            self._test_mode = test_mode
        self._debug_mode = debug_mode
        self._dlc_base = dlc_base
        self._moseq_base = moseq_base

        if not debug_mode:
            base_path.mkdir(parents=True, exist_ok=True)
        Path(self._dlc_base).mkdir(parents=True, exist_ok=True)
        Path(self._moseq_base).mkdir(parents=True, exist_ok=True)

        loaded_env = self._load_env_vars()
        self._set_env_with_dict(
            {**config_dirs, **kachery_zone_dict, **loaded_env}
        )
        self._mkdirs_from_dict_vals(config_dirs)

        self._config = dict(
            debug_mode=self._debug_mode,
            test_mode=self.test_mode,
            **self.config_defaults,
            **config_dirs,
            **kachery_zone_dict,
            **loaded_env,
        )

        self._set_dj_config_stores()

        self.load_failed = False

        return self._config

    def _load_env_vars(self) -> dict:
        loaded_dict = {}
        for var, val in self.env_defaults.items():
            loaded_dict[var] = os.getenv(var, val)
        return loaded_dict

    def _set_env_with_dict(self, env_dict) -> None:
        # NOTE: Kept for backwards compatibility. Should be removed in future
        # for custom paths. Keep self.env_defaults.
        # SPYGLASS_BASE_DIR may be used for docker assembly of export
        for var, val in env_dict.items():
            os.environ[var] = str(val)

    def _mkdirs_from_dict_vals(self, dir_dict) -> None:
        if self._debug_mode:
            return
        for dir_str in dir_dict.values():
            Path(dir_str).mkdir(parents=True, exist_ok=True)

    def _set_dj_config_stores(self, check_match=True, set_stores=True) -> None:
        """
        Checks dj.config['stores'] match resolved dirs. Ensures stores set.

        Parameters
        ----------
        check_match: bool
            Optional. Default True. Check that dj.config['stores'] match
            resolved dirs.
        set_stores: bool
            Optional. Default True. Set dj.config['stores'] to resolved dirs.
        """

        mismatch_analysis = False
        mismatch_raw = False

        if check_match:
            dj_stores = dj.config.get("stores", {})
            store_r = dj_stores.get("raw", {}).get("location")
            store_a = dj_stores.get("analysis", {}).get("location")
            mismatch_raw = store_r and Path(store_r) != Path(self.raw_dir)
            mismatch_analysis = store_a and Path(store_a) != Path(
                self.analysis_dir
            )

        if set_stores:
            if (mismatch_raw or mismatch_analysis) and not self.test_mode:
                logger.warning(
                    "Setting config DJ stores to resolve mismatch.\n\t"
                    + f"raw     : {self.raw_dir}\n\t"
                    + f"analysis: {self.analysis_dir}"
                )
            dj.config.update(self._dj_stores)
            return

        if mismatch_raw or mismatch_analysis:
            raise ValueError(
                "dj.config['stores'] does not match resolved dirs."
                + f"\n\tdj.config['stores']: {dj_stores}"
                + f"\n\tResolved dirs: {self._dj_stores}"
            )

        return

    def dir_to_var(self, dir: str, dir_type: str = "spyglass") -> str:
        """Converts a dir string to an env variable name."""
        return f"{dir_type.upper()}_{dir.upper()}_DIR"

    def _generate_dj_config(
        self,
        base_dir: str = None,
        database_user: str = None,
        database_password: str = None,
        database_host: str = "lmf-db.cin.ucsf.edu",
        database_port: int = 3306,
        database_use_tls: bool = True,
        **kwargs,
    ) -> dict:
        """Generate a datajoint configuration file.

        Parameters
        ----------
        base_dir : str, optional
            The base directory. If not provided, will use existing config.
        database_user : str, optional
            The database user. If not provided, resulting config will not
            specify.
        database_password : str, optional
            The database password. If not provided, resulting config will not
            specify.
        database_host : str, optional
            Default lmf-db.cin.ucsf.edu. MySQL host name.
        database_port : int, optional
            Default 3306. Port number for MySQL server.
        database_use_tls : bool, optional
            Default True. Use TLS encryption.
        **kwargs: dict, optional
            Any other valid datajoint configuration parameters.
            Note: python will raise error for params with `.` in name.
        """

        if database_user:
            kwargs.update({"database.user": database_user})
        if database_password:
            kwargs.update({"database.password": database_password})

        kwargs.update(
            {
                "database.host": database_host,
                "database.port": database_port,
                "database.use_tls": database_use_tls,
            }
        )

        # `|` merges dictionaries
        return self.dj_defaults | self._dj_stores | self._dj_custom | kwargs

    def save_dj_config(
        self,
        save_method: str = "global",
        output_filename: str = None,
        base_dir=None,
        set_password=True,
        **kwargs,
    ) -> None:
        """Set the dj.config parameters, set password, and save config to file.

        Parameters
        ----------
        save_method : {'local', 'global', 'custom'}, optional
            The method to use to save the config. If either 'local' or 'global',
            datajoint builtins will be used to save.
        output_filename : str or Path, optional
            Default to datajoint global config. If save_method = 'custom', name
            of file to generate. Must end in either yaml or json.
        base_dir : str, optional
            The base directory. If not provided, will default to the env var
        set_password : bool, optional
            Default True. Set the database password.
        kwargs: dict, optional
            Any other valid datajoint configuration parameters, including
            database_user, database_password, database_host, database_port, etc.
            Note: python will raise error for params with `.` in name, so use
            underscores instead.
        """
        if base_dir:
            self.load_config(
                base_dir=base_dir, force_reload=True, on_startup=False
            )

        if output_filename:
            save_method = "custom"
            path = Path(output_filename).expanduser()  # Expand ~
            filepath = (
                path if path.is_absolute() else path.resolve()
            )  # Resolve relative paths and symlinks
            filepath.parent.mkdir(exist_ok=True, parents=True)
            filepath = (
                filepath.with_suffix(".json")  # ensure suffix, default json
                if filepath.suffix not in [".json", ".yaml"]
                else filepath
            )
        elif save_method == "local":
            filepath = Path(".") / dj.settings.LOCALCONFIG
        elif save_method == "global":
            filepath = Path("~").expanduser() / dj.settings.GLOBALCONFIG
        else:
            raise ValueError(
                "For save_dj_config, either (a) save_method must be 'local' "
                + " or 'global' or (b) must provide custom output_filename."
            )

        dj.config.update(self._generate_dj_config(**kwargs))

        if set_password:
            try:
                dj.set_password()
            except OperationalError as e:
                warnings.warn(f"Database connection issues. Wrong pass?\n\t{e}")

        user_warn = (
            f"Replace existing file? {filepath.resolve()}\n\t"
            + "\n\t".join(
                [
                    f"{k}: {v if k != 'database.password' else '***'}"
                    for k, v in dj.config.items()
                ]
            )
            + "\n"
        )

        if (
            not self.test_mode
            and filepath.exists()
            and dj.utils.user_choice(user_warn)[0] != "y"
        ):
            return

        if save_method == "global":
            dj.config.save_global(verbose=True)
            return

        if save_method == "local":
            dj.config.save_local(verbose=True)
            return

        with open(filepath, "w") as outfile:
            if filepath.suffix == ".yaml":
                yaml.dump(dj.config._conf, outfile, default_flow_style=False)
            else:
                json.dump(dj.config._conf, outfile, indent=2)
            logger.info(f"Saved config to {filepath}")

    @property
    def _dj_stores(self) -> dict:
        self.load_config()
        return {
            "stores": {
                "raw": {
                    "protocol": "file",
                    "location": self.raw_dir,
                    "stage": self.raw_dir,
                },
                "analysis": {
                    "protocol": "file",
                    "location": self.analysis_dir,
                    "stage": self.analysis_dir,
                },
            }
        }

    @property
    def _dj_custom(self) -> dict:
        self.load_config()
        return {
            "custom": {
                "debug_mode": str(self.debug_mode).lower(),
                "test_mode": str(self.test_mode).lower(),
                "spyglass_dirs": {
                    "base": self.base_dir,
                    "raw": self.raw_dir,
                    "analysis": self.analysis_dir,
                    "recording": self.recording_dir,
                    "sorting": self.sorting_dir,
                    "waveforms": self.waveforms_dir,
                    "temp": self.temp_dir,
                    "video": self.video_dir,
                    "export": self.export_dir,
                },
                "kachery_dirs": {
                    "cloud": self.config.get(
                        self.dir_to_var("cloud", "kachery")
                    ),
                    "storage": self.config.get(
                        self.dir_to_var("storage", "kachery")
                    ),
                    "temp": self.config.get(self.dir_to_var("temp", "kachery")),
                },
                "dlc_dirs": {
                    "base": self._dlc_base,
                    "project": self.dlc_project_dir,
                    "video": self.dlc_video_dir,
                    "output": self.dlc_output_dir,
                },
                "moseq_dirs": {
                    "base": self._moseq_base,
                    "project": self.moseq_project_dir,
                    "video": self.moseq_video_dir,
                },
                "kachery_zone": os.environ.get(
                    "KACHERY_ZONE", "franklab.default"
                ),
            }
        }

    @property
    def config(self) -> dict:
        """Dictionary of config settings."""
        self.load_config()
        return self._config

    @property
    def base_dir(self) -> str:
        """Base directory as a string."""
        return self.config.get(self.dir_to_var("base"))

    @property
    def raw_dir(self) -> str:
        """Raw data directory as a string."""
        return self.config.get(self.dir_to_var("raw"))

    @property
    def analysis_dir(self) -> str:
        """Analysis directory as a string."""
        return self.config.get(self.dir_to_var("analysis"))

    @property
    def recording_dir(self) -> str:
        """Recording directory as a string."""
        return self.config.get(self.dir_to_var("recording"))

    @property
    def sorting_dir(self) -> str:
        """Sorting directory as a string."""
        return self.config.get(self.dir_to_var("sorting"))

    @property
    def waveforms_dir(self) -> str:
        """Waveforms directory as a string."""
        return self.config.get(self.dir_to_var("waveforms"))

    @property
    def temp_dir(self) -> str:
        """Temp directory as a string."""
        return self.config.get(self.dir_to_var("temp"))

    @property
    def video_dir(self) -> str:
        """Video directory as a string."""
        return self.config.get(self.dir_to_var("video"))

    @property
    def export_dir(self) -> str:
        """Export directory as a string."""
        return self.config.get(self.dir_to_var("export"))

    @property
    def debug_mode(self) -> bool:
        """Returns True if debug_mode is set.

        Supports skipping inserts for Dockerized development.
        """
        return self._debug_mode

    @property
    def test_mode(self) -> bool:
        """Returns True if test_mode is set.

        Required for pytests to run without prompts."""
        if self._test_mode is not _UNSET:
            return self._test_mode
        if self._initial_test_mode is not _UNSET:
            return str_to_bool(self._initial_test_mode)
        return False

    @property
    def dlc_project_dir(self) -> str:
        """DLC project directory as a string."""
        return self.config.get(self.dir_to_var("project", "dlc"))

    @property
    def dlc_video_dir(self) -> str:
        """DLC video directory as a string."""
        return self.config.get(self.dir_to_var("video", "dlc"))

    @property
    def dlc_output_dir(self) -> str:
        """DLC output directory as a string."""
        return self.config.get(self.dir_to_var("output", "dlc"))

    @property
    def moseq_project_dir(self) -> str:
        """Moseq project directory as a string."""
        return self.config.get(self.dir_to_var("project", "moseq"))

    @property
    def moseq_video_dir(self) -> str:
        """Moseq video directory as a string."""
        return self.config.get(self.dir_to_var("video", "moseq"))

__init__(base_dir=None, **kwargs)

Initializes a new instance of the class.

Parameters:

Name	Type	Description	Default
base_dir	str	The base directory.	None

Attributes:

Name	Type	Description
supplied_base_dir (str)		The base directory passed to the class.
config_defaults (dict)		Default settings for the config.
relative_dirs (dict)		Relative dirs for each prefix (spyglass, kachery, dlc). Relative to respective base_dir. Created on init.
dj_defaults (dict)		Default settings for datajoint.
env_defaults (dict)		Default settings for environment variables.
_config (dict)		Cached config settings.
_debug_mode (bool)		True if debug_mode is set. Supports skipping known bugs in test env.
_test_mode (bool or object)		The bound test mode, or _UNSET before the first deliberate or successful load. The public test_mode property always returns a bool.

Source code in src/spyglass/settings.py

def __init__(self, base_dir: str = None, **kwargs) -> None:
    """
    Initializes a new instance of the class.

    Parameters
    ----------
    base_dir (str)
        The base directory.

    Attributes
    ----------
    supplied_base_dir (str)
        The base directory passed to the class.
    config_defaults (dict)
        Default settings for the config.
    relative_dirs (dict)
        Relative dirs for each prefix (spyglass, kachery, dlc). Relative
        to respective base_dir. Created on init.
    dj_defaults (dict)
        Default settings for datajoint.
    env_defaults (dict)
        Default settings for environment variables.
    _config (dict)
        Cached config settings.
    _debug_mode (bool)
        True if debug_mode is set. Supports skipping known bugs in test env.
    _test_mode (bool or object)
        The bound test mode, or ``_UNSET`` before the first deliberate or
        successful load. The public ``test_mode`` property always returns
        a bool.
    """
    self.supplied_base_dir = base_dir
    self._config = dict()
    self.config_defaults = dict(prepopulate=True)
    # Constructor values participate in first-load precedence. test_mode
    # becomes instance identity once an explicit load starts or an ambient
    # load succeeds; debug_mode remains an ordinary reloadable setting.
    self._debug_mode_arg = kwargs.get("debug_mode", _UNSET)
    self._initial_test_mode = kwargs.get("test_mode", _UNSET)
    self._debug_mode = (
        False
        if self._debug_mode_arg is _UNSET
        else str_to_bool(self._debug_mode_arg)
    )
    self._test_mode = _UNSET
    self._dlc_base = None
    # Initialized here, not only in load_config's COMMIT phase: a load
    # that fails or returns early (e.g. no base under an ambient test
    # mode) still leaves `_dj_custom`/`_generate_dj_config` able to read
    # it, matching `_dlc_base`.
    self._moseq_base = None
    self.load_failed = False
    # A mode-change request invalidates a loaded instance permanently. Keep
    # only the message (not an exception/traceback); recovery uses a new
    # SpyglassConfig object with an unambiguous lifecycle.
    self._mode_error: str | None = None

    # Load directory schema from JSON file (single source of truth)
    # {PREFIX}_{KEY}_DIR, default dir relative to base_dir
    # NOTE: Adding new dir requires edit to HHMI hub AND directory_schema.json
    self.relative_dirs = self._load_directory_schema()
    self.dj_defaults = {
        "database.host": kwargs.get("database_host", "lmf-db.cin.ucsf.edu"),
        "database.user": kwargs.get("database_user"),
        "database.port": kwargs.get("database_port", 3306),
        "database.use_tls": kwargs.get("database_use_tls", True),
        "filepath_checksum_size_limit": 1 * 1024**3,
        "enable_python_native_blobs": True,
    }
    self.env_defaults = {
        "FIGURL_CHANNEL": "franklab2",
        "DJ_SUPPORT_FILEPATH_MANAGEMENT": "TRUE",
        "KACHERY_CLOUD_EPHEMERAL": "TRUE",
        "HDF5_USE_FILE_LOCKING": "FALSE",
    }

load_config(base_dir=None, force_reload=False, on_startup=False, **kwargs)

Loads the configuration settings for the object.

Order of precedence, where X is base, raw, analysis, etc.: 1. SpyglassConfig(base_dir="string") for base dir only 2. dj.config['custom']['spyglass_dirs']['X'] 3. dj.config['custom']['kachery_dirs']['X'] 4. os.environ['{SPYGLASS/KACHERY}_{X}_DIR'] 5. resolved_base_dir/X for non-base dirs

When test_mode=True, environment variables are not consulted for any directory path, and the resolved base_dir must contain a 'tests' path component.

test_mode binds to the instance on the first deliberate or successful load and is then immutable. Passing a test_mode that differs from the bound value (even without force_reload) does not transition the instance -- it invalidates it and raises. To switch modes, construct a new SpyglassConfig.

Parameters:

Name	Type	Description	Default
base_dir		Optional. Default None. The base directory. If not provided, will use the env variable or existing config.	None
force_reload		Optional. Default False. Default skip load if already completed.	False

Raises:

Type	Description
ValueError	When a caller attempts to change the mode of a bound instance; or, under test_mode, when a deliberate load cannot resolve a base_dir, the resolved base_dir does not contain a 'tests' path component, or any resolved directory -- including one reached through a symlink -- falls outside that base_dir. A fresh ambient (dj.config-only) test-mode load with no base returns gracefully instead.

Returns:

Type	Description
dict	list of relative_dirs and other settings (e.g., prepopulate).

Source code in src/spyglass/settings.py

def load_config(
    self,
    base_dir=None,
    force_reload=False,
    on_startup: bool = False,
    **kwargs,
) -> dict | None:
    """
    Loads the configuration settings for the object.

    Order of precedence, where X is base, raw, analysis, etc.:
    1. SpyglassConfig(base_dir="string") for base dir only
    2. dj.config['custom']['spyglass_dirs']['X']
    3. dj.config['custom']['kachery_dirs']['X']
    4. os.environ['{SPYGLASS/KACHERY}_{X}_DIR']
    5. resolved_base_dir/X for non-base dirs

    When test_mode=True, environment variables are not consulted for any
    directory path, and the resolved base_dir must contain a 'tests' path
    component.

    ``test_mode`` binds to the instance on the first deliberate or
    successful load and is then immutable. Passing a ``test_mode`` that
    differs from the bound value (even without ``force_reload``) does not
    transition the instance -- it invalidates it and raises. To switch
    modes, construct a new ``SpyglassConfig``.

    Parameters
    ----------
    base_dir: str
        Optional. Default None. The base directory. If not provided, will
        use the env variable or existing config.
    force_reload: bool
        Optional. Default False. Default skip load if already completed.

    Raises
    ------
    ValueError
        When a caller attempts to change the mode of a bound instance; or,
        under test_mode, when a deliberate load cannot resolve a base_dir,
        the resolved base_dir does not contain a 'tests' path component, or
        any resolved directory -- including one reached through a symlink
        -- falls outside that base_dir. A fresh ambient (dj.config-only)
        test-mode load with no base returns gracefully instead.

    Returns
    -------
    dict
        list of relative_dirs and other settings (e.g., prepopulate).
    """
    # Fast path for the common cached read (every directory property
    # routes here with no kwargs). A mode-change request (explicit
    # test_mode=) still falls through to _resolve_test_mode's binding /
    # rejection, and a mode-wedged instance has _config == {} (falsy) so it
    # also falls through and re-raises.
    if (
        not force_reload
        and self._config
        and kwargs.get("test_mode", _UNSET) is _UNSET
    ):
        return self._config

    dj_custom = dj.config.get("custom", {})
    dj_spyglass = dj_custom.get("spyglass_dirs", {})
    dj_kachery = dj_custom.get("kachery_dirs", {})
    dj_dlc = dj_custom.get("dlc_dirs", {})
    dj_moseq = dj_custom.get("moseq_dirs", {})

    test_mode, test_mode_is_bound = self._resolve_test_mode(
        kwargs.get("test_mode", _UNSET), dj_custom
    )
    if not force_reload and self._config:
        return self._config

    def _resolve_debug_mode() -> bool:
        """Resolve call > constructor > DataJoint > default precedence."""
        call_value = kwargs.get("debug_mode", _UNSET)
        if call_value is not _UNSET:
            return str_to_bool(call_value)
        if self._debug_mode_arg is not _UNSET:
            return str_to_bool(self._debug_mode_arg)
        return str_to_bool(dj_custom.get("debug_mode", False))

    debug_mode = _resolve_debug_mode()

    # Until a deliberate test-mode load commits, keep the object visibly
    # failed. A successful commit below resets this flag. Same-mode reloads
    # of an existing valid test config remain transactional.
    if test_mode and test_mode_is_bound and not self._config:
        self.load_failed = True

    resolved_base = (
        base_dir
        or self.supplied_base_dir
        or dj_spyglass.get("base")
        # Gated by test_mode like every other directory env var below:
        # SPYGLASS_BASE_DIR is the exact production path this sandbox
        # exists to keep destructive tests off, so test_mode must not
        # inherit it. Explicit base_dir/config still resolve.
        or (None if test_mode else os.environ.get("SPYGLASS_BASE_DIR"))
    )

    # Log when supplied base_dir causes environment variable overrides to be ignored
    if self.supplied_base_dir:
        logger.info(
            "Using supplied base_dir - ignoring SPYGLASS_* environment variable overrides"
        )

    # ---------------------------- RESOLVE ----------------------------
    # Compute every path as a plain value. Nothing is created and no
    # external/global state is mutated until validation passes.
    if not resolved_base:
        self.load_failed = True
        if test_mode and test_mode_is_bound:
            raise ValueError(
                "Refusing to load Spyglass in test_mode without an "
                "explicit base_dir or "
                "dj.config['custom']['spyglass_dirs']['base']; "
                "SPYGLASS_BASE_DIR is ignored in test_mode."
            )
        if not on_startup:  # Only warn if not on startup
            logger.error(
                "Could not find SPYGLASS_BASE_DIR"
                + "\n\tCheck dj.config['custom']['spyglass_dirs']['base']"
                + "\n\tand os.environ['SPYGLASS_BASE_DIR']"
            )
        return

    base_path = Path(resolved_base).expanduser().resolve()
    resolved_base = str(base_path)

    def env_or_none(var: str) -> str | None:
        """Read an env var, ignored in test_mode to keep the sandbox."""
        return None if test_mode else os.environ.get(var)

    dlc_project = env_or_none("DLC_PROJECT_PATH")
    dlc_base = (
        dj_dlc.get("base")
        or env_or_none("DLC_BASE_DIR")
        or (dlc_project.split("projects")[0] if dlc_project else None)
        or str(Path(resolved_base) / "deeplabcut")
    )
    moseq_base = (
        dj_moseq.get("base")
        or env_or_none("MOSEQ_BASE_DIR")
        or str(Path(resolved_base) / "moseq")
    )

    config_dirs = {"SPYGLASS_BASE_DIR": str(resolved_base)}
    source_config_lookup = {
        "dlc": dj_dlc,
        "moseq": dj_moseq,
        "kachery": dj_kachery,
    }
    base_lookup = {"dlc": dlc_base, "moseq": moseq_base}
    for prefix, dirs in self.relative_dirs.items():
        this_base = base_lookup.get(prefix, resolved_base)
        for dir, dir_str in dirs.items():
            dir_env_fmt = self.dir_to_var(dir=dir, dir_type=prefix)

            env_loc = (  # Ignore env vars if base was passed or test_mode
                os.environ.get(dir_env_fmt)
                if not self.supplied_base_dir and not test_mode
                else None
            )
            source_config = source_config_lookup.get(prefix, dj_spyglass)
            dir_location = (
                source_config.get(dir)
                or env_loc
                or str(Path(this_base) / dir_str)
            ).replace('"', "")

            config_dirs.update({dir_env_fmt: str(dir_location)})

    kachery_zone_dict = {
        "KACHERY_ZONE": (
            env_or_none("KACHERY_ZONE")
            or dj.config.get("custom", {}).get("kachery_zone")
            or "franklab.default"
        )
    }

    # ---------------------------- VALIDATE ---------------------------
    # Both checks apply ONLY under test_mode. Production configuration
    # is unchanged: an analysis dir anywhere, including behind a
    # symlink, stays legal.
    if test_mode:
        validation_error = None
        if "tests" not in base_path.parts:
            validation_error = (
                f"Refusing to load Spyglass in test_mode with base_dir "
                f"{resolved_base!r}: path does not contain a 'tests' "
                "component. Run pytest with --base-dir pointing inside a "
                "tests/ directory (default: ./tests/_data/) to keep "
                "destructive operations off shared/production storage."
            )
        else:
            # Path.resolve() is non-strict: a dir that does not exist yet
            # resolves to its would-be path, while an EXISTING symlink
            # resolves through to its target. That is what catches an
            # analysis dir symlinked at production storage.
            checked = dict(config_dirs)
            checked["DLC_BASE_DIR"] = dlc_base
            checked["MOSEQ_BASE_DIR"] = moseq_base
            for var, loc in checked.items():
                loc_path = Path(loc).expanduser().resolve()
                if not loc_path.is_relative_to(base_path):
                    validation_error = (
                        f"Refusing to load Spyglass in test_mode: {var} "
                        f"resolves to {str(loc_path)!r}, outside the test "
                        f"base {resolved_base!r}. Destructive tests must "
                        "stay within the test base directory; check "
                        "dj.config['custom'] and any directory symlinks."
                    )
                    break

        if validation_error is not None:
            # A deliberate (bound) test-mode load fails loud. An ambient /
            # implicit load must not crash unrelated code -- matching the
            # no-base handling above -- but must also NOT commit a
            # test-mode config whose paths escape the sandbox. So mark the
            # load failed and return without committing, leaving the mode
            # unbound.
            self.load_failed = True
            if test_mode_is_bound:
                raise ValueError(validation_error)
            if not on_startup:  # Only warn if not on startup
                logger.error(validation_error)
            return

    # ----------------------------- COMMIT ----------------------------
    if self._test_mode is _UNSET:
        self._test_mode = test_mode
    self._debug_mode = debug_mode
    self._dlc_base = dlc_base
    self._moseq_base = moseq_base

    if not debug_mode:
        base_path.mkdir(parents=True, exist_ok=True)
    Path(self._dlc_base).mkdir(parents=True, exist_ok=True)
    Path(self._moseq_base).mkdir(parents=True, exist_ok=True)

    loaded_env = self._load_env_vars()
    self._set_env_with_dict(
        {**config_dirs, **kachery_zone_dict, **loaded_env}
    )
    self._mkdirs_from_dict_vals(config_dirs)

    self._config = dict(
        debug_mode=self._debug_mode,
        test_mode=self.test_mode,
        **self.config_defaults,
        **config_dirs,
        **kachery_zone_dict,
        **loaded_env,
    )

    self._set_dj_config_stores()

    self.load_failed = False

    return self._config

dir_to_var(dir, dir_type='spyglass')

Converts a dir string to an env variable name.

Source code in src/spyglass/settings.py

def dir_to_var(self, dir: str, dir_type: str = "spyglass") -> str:
    """Converts a dir string to an env variable name."""
    return f"{dir_type.upper()}_{dir.upper()}_DIR"

save_dj_config(save_method='global', output_filename=None, base_dir=None, set_password=True, **kwargs)

Set the dj.config parameters, set password, and save config to file.

Parameters:

Name	Type	Description	Default
save_method	(local, 'global', custom)	The method to use to save the config. If either 'local' or 'global', datajoint builtins will be used to save.	'local'
output_filename	str or Path	Default to datajoint global config. If save_method = 'custom', name of file to generate. Must end in either yaml or json.	None
base_dir	str	The base directory. If not provided, will default to the env var	None
set_password	bool	Default True. Set the database password.	True
kwargs		Any other valid datajoint configuration parameters, including database_user, database_password, database_host, database_port, etc. Note: python will raise error for params with . in name, so use underscores instead.	{}

Source code in src/spyglass/settings.py

def save_dj_config(
    self,
    save_method: str = "global",
    output_filename: str = None,
    base_dir=None,
    set_password=True,
    **kwargs,
) -> None:
    """Set the dj.config parameters, set password, and save config to file.

    Parameters
    ----------
    save_method : {'local', 'global', 'custom'}, optional
        The method to use to save the config. If either 'local' or 'global',
        datajoint builtins will be used to save.
    output_filename : str or Path, optional
        Default to datajoint global config. If save_method = 'custom', name
        of file to generate. Must end in either yaml or json.
    base_dir : str, optional
        The base directory. If not provided, will default to the env var
    set_password : bool, optional
        Default True. Set the database password.
    kwargs: dict, optional
        Any other valid datajoint configuration parameters, including
        database_user, database_password, database_host, database_port, etc.
        Note: python will raise error for params with `.` in name, so use
        underscores instead.
    """
    if base_dir:
        self.load_config(
            base_dir=base_dir, force_reload=True, on_startup=False
        )

    if output_filename:
        save_method = "custom"
        path = Path(output_filename).expanduser()  # Expand ~
        filepath = (
            path if path.is_absolute() else path.resolve()
        )  # Resolve relative paths and symlinks
        filepath.parent.mkdir(exist_ok=True, parents=True)
        filepath = (
            filepath.with_suffix(".json")  # ensure suffix, default json
            if filepath.suffix not in [".json", ".yaml"]
            else filepath
        )
    elif save_method == "local":
        filepath = Path(".") / dj.settings.LOCALCONFIG
    elif save_method == "global":
        filepath = Path("~").expanduser() / dj.settings.GLOBALCONFIG
    else:
        raise ValueError(
            "For save_dj_config, either (a) save_method must be 'local' "
            + " or 'global' or (b) must provide custom output_filename."
        )

    dj.config.update(self._generate_dj_config(**kwargs))

    if set_password:
        try:
            dj.set_password()
        except OperationalError as e:
            warnings.warn(f"Database connection issues. Wrong pass?\n\t{e}")

    user_warn = (
        f"Replace existing file? {filepath.resolve()}\n\t"
        + "\n\t".join(
            [
                f"{k}: {v if k != 'database.password' else '***'}"
                for k, v in dj.config.items()
            ]
        )
        + "\n"
    )

    if (
        not self.test_mode
        and filepath.exists()
        and dj.utils.user_choice(user_warn)[0] != "y"
    ):
        return

    if save_method == "global":
        dj.config.save_global(verbose=True)
        return

    if save_method == "local":
        dj.config.save_local(verbose=True)
        return

    with open(filepath, "w") as outfile:
        if filepath.suffix == ".yaml":
            yaml.dump(dj.config._conf, outfile, default_flow_style=False)
        else:
            json.dump(dj.config._conf, outfile, indent=2)
        logger.info(f"Saved config to {filepath}")

config property

Dictionary of config settings.

base_dir property

Base directory as a string.

raw_dir property

Raw data directory as a string.

analysis_dir property

Analysis directory as a string.

recording_dir property

Recording directory as a string.

sorting_dir property

Sorting directory as a string.

waveforms_dir property

Waveforms directory as a string.

temp_dir property

Temp directory as a string.

video_dir property

Video directory as a string.

export_dir property

Export directory as a string.

debug_mode property

Returns True if debug_mode is set.

Supports skipping inserts for Dockerized development.

test_mode property

Returns True if test_mode is set.

Required for pytests to run without prompts.

dlc_project_dir property

DLC project directory as a string.

dlc_video_dir property

DLC video directory as a string.

dlc_output_dir property

DLC output directory as a string.

moseq_project_dir property

Moseq project directory as a string.

moseq_video_dir property

Moseq video directory as a string.