apps.open_ephys

OpenEphysAppSettings

Bases: ServiceSettings

Settings for Open Ephys App.

settings_customise_sources classmethod

settings_customise_sources(
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]

Customizes the settings sources to include the safe YAML settings source.

Parameters:

Name	Type	Description	Default
settings_cls	type[BaseSettings]	The settings class	required
init_settings	PydanticBaseSettingsSource	The initial settings source	required
env_settings	PydanticBaseSettingsSource	The environment settings source	required
dotenv_settings	PydanticBaseSettingsSource	The dotenv settings source	required
file_secret_settings	PydanticBaseSettingsSource	The file secret settings source	required

Returns:

Type	Description
tuple[PydanticBaseSettingsSource, ...]	Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources

Source code in src/clabe/services.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[ps.BaseSettings],
    init_settings: ps.PydanticBaseSettingsSource,
    env_settings: ps.PydanticBaseSettingsSource,
    dotenv_settings: ps.PydanticBaseSettingsSource,
    file_secret_settings: ps.PydanticBaseSettingsSource,
) -> tuple[ps.PydanticBaseSettingsSource, ...]:
    """
    Customizes the settings sources to include the safe YAML settings source.

    Args:
        settings_cls: The settings class
        init_settings: The initial settings source
        env_settings: The environment settings source
        dotenv_settings: The dotenv settings source
        file_secret_settings: The file secret settings source

    Returns:
        Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources
    """
    yaml_sources = [
        _SafeYamlSettingsSource(settings_cls, yaml_file=p, yaml_config_section=cls.__yml_section__)
        for p in KNOWN_CONFIG_FILES
    ]
    # The in-memory document ranks right after the first (local override) config file
    return (
        init_settings,
        *yaml_sources[:1],
        _InMemorySettingsSource(settings_cls, config_section=cls.__yml_section__),
        *yaml_sources[1:],
        env_settings,
        dotenv_settings,
        file_secret_settings,
    )

OpenEphysApp

OpenEphysApp(
    settings: OpenEphysAppSettings,
    skip_validation: bool = False,
)

Bases: ExecutableApp, _DefaultExecutorMixin

A class to manage the execution of Open Ephys GUI.

Handles Open Ephys GUI execution, configuration management, and process monitoring for ephys experiments.

Methods:

Name	Description
run	Executes the Open Ephys GUI
get_result	Retrieves the result of the Open Ephys execution
add_app_settings	Adds or updates application settings
validate	Validates the Open Ephys application configuration

Initializes the OpenEphysApp instance.

Parameters:

Name	Type	Description	Default
settings	OpenEphysAppSettings	Settings for the Open Ephys App	required
skip_validation	bool	Whether to skip validation of required files	False

Example

# Create and run a Open Ephys app
app = OpenEphysApp(settings=OpenEphysAppSettings(signal_chain="signal_chain.xml"))
app.run()

Source code in src/clabe/apps/open_ephys.py

def __init__(
    self,
    settings: OpenEphysAppSettings,
    skip_validation: bool = False,
) -> None:
    """
    Initializes the OpenEphysApp instance.

    Args:
        settings: Settings for the Open Ephys App
        skip_validation: Whether to skip validation of required files

    Example:
        ```python
        # Create and run a Open Ephys app
        app = OpenEphysApp(settings=OpenEphysAppSettings(signal_chain="signal_chain.xml"))
        app.run()
        ```
    """
    self.settings = settings
    self.signal_chain = Path(self.settings.signal_chain).resolve()
    self.executable = Path(self.settings.executable).resolve()
    self._client = _OpenEphysGuiClient(host=self.settings.address, port=self.settings.port)

    if not skip_validation:
        self.validate()

    self._command = Command[CommandResult](
        cmd=[str(self.executable), str(self.signal_chain)], output_parser=identity_parser
    )

command property

command: Command[CommandResult]

Get the command to execute.

client property

client: _OpenEphysGuiClient

Get the Open Ephys GUI client.

validate

validate()

Validates the existence of required files and directories.

Raises:

Type	Description
FileNotFoundError	If any required file or directory is missing

Source code in src/clabe/apps/open_ephys.py

def validate(self):
    """
    Validates the existence of required files and directories.

    Raises:
        FileNotFoundError: If any required file or directory is missing
    """
    if not Path(self.executable).exists():
        raise FileNotFoundError(f"Executable not found: {self.executable}")
    if not Path(self.signal_chain).exists():
        raise FileNotFoundError(f"Signal chain file not found: {self.signal_chain}")

run

run(
    executor_kwargs: dict[str, Any] | None = None,
) -> CommandResult

Execute the command using a local executor and return the result.

Parameters:

Name	Type	Description	Default
executor_kwargs	dict[str, Any] | None	Keyword arguments forwarded to the local executor.	None

Source code in src/clabe/apps/_executors.py

@runnable
def run(self, executor_kwargs: dict[str, Any] | None = None) -> CommandResult:
    """Execute the command using a local executor and return the result.

    Args:
        executor_kwargs: Keyword arguments forwarded to the local executor.
    """
    return self.command.execute(LocalExecutor(**(executor_kwargs or {})))

run_async async

run_async(
    executor_kwargs: dict[str, Any] | None = None,
) -> CommandResult

Execute the command asynchronously using a local executor and return the result.

Parameters:

Name	Type	Description	Default
executor_kwargs	dict[str, Any] | None	Keyword arguments forwarded to the local executor.	None

Source code in src/clabe/apps/_executors.py

@runnable
async def run_async(self, executor_kwargs: dict[str, Any] | None = None) -> CommandResult:
    """Execute the command asynchronously using a local executor and return the result.

    Args:
        executor_kwargs: Keyword arguments forwarded to the local executor.
    """
    return await self.command.execute_async(AsyncLocalExecutor(**(executor_kwargs or {})))

Status

Bases: str, Enum

GUI acquisition/recording mode.

StatusResponse

Bases: BaseModel

Response from /api/status endpoint.

StatusRequest

Bases: BaseModel

Request to set GUI acquisition/recording mode.

RecordNode

Bases: BaseModel

Information about a Record Node.

RecordingResponse

Bases: BaseModel

Response from /api/recording endpoint.

RecordingRequest

Bases: BaseModel

Request to update recording configuration.

RecordNodeRequest

Bases: BaseModel

Request to update a specific Record Node.

Stream

Bases: BaseModel

Data stream information.

Processor

Bases: BaseModel

Processor/plugin information.

ProcessorsResponse

Bases: BaseModel

Response from /api/processors endpoint.

ConfigRequest

Bases: BaseModel

Request to send configuration message to a processor.

MessageRequest

Bases: BaseModel

Request to broadcast a message to all processors.

WindowRequest

Bases: BaseModel

Request to control GUI window.