IO Loaders

The slopit.io module provides loaders for various data formats. Loaders convert platform-specific formats into the standardized SlopitSession schema.

High-Level Functions

load_session

Load a single session from a file with automatic format detection.

slopit.io.load_session

load_session(path: str | Path) -> SlopitSession

Load a single session from a file.

Automatically detects the file format and uses the appropriate loader.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the session file.	required

Returns:

Type	Description
SlopitSession	The loaded session data.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.
ValueError	If the file format cannot be determined or is invalid.

Examples:

>>> session = load_session("data/participant_001.json")
>>> print(f"Session {session.session_id} has {len(session.trials)} trials")

Example

from slopit import load_session

# Load native slopit format
session = load_session("data/participant_001.json")

# Load JATOS format
session = load_session("jatos_results/study_result_123.txt")

load_sessions

Load multiple sessions from a directory.

slopit.io.load_sessions

load_sessions(path: str | Path, pattern: str = '*') -> list[SlopitSession]

Load multiple sessions from a directory.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to directory containing session files.	required
pattern	str	Glob pattern for file matching.	'*'

Returns:

Type	Description
list[SlopitSession]	List of loaded sessions.

Examples:

>>> sessions = load_sessions("data/")
>>> print(f"Loaded {len(sessions)} sessions")

Example

from slopit import load_sessions

# Load all sessions from a directory
sessions = load_sessions("data/")
print(f"Loaded {len(sessions)} sessions")

# Filter by pattern
sessions = load_sessions("data/", pattern="*.json")

Base Loader Class

BaseLoader

Abstract base class that all format-specific loaders inherit from.

slopit.io.base.BaseLoader

Bases: ABC

Abstract base class for data loaders.

Subclasses implement loading logic for specific data formats (JATOS, Pavlovia, Gorilla, etc.).

Source code in src/slopit/io/base.py

class BaseLoader(ABC):
    """Abstract base class for data loaders.

    Subclasses implement loading logic for specific data formats
    (JATOS, Pavlovia, Gorilla, etc.).
    """

    @abstractmethod
    def load(self, path: Path) -> SlopitSession:
        """Load a single session from a file.

        Parameters
        ----------
        path
            Path to the data file.

        Returns
        -------
        SlopitSession
            The loaded session data.

        Raises
        ------
        FileNotFoundError
            If the file does not exist.
        ValueError
            If the file format is invalid.
        """
        ...

    @abstractmethod
    def load_many(self, path: Path, pattern: str = "*") -> Iterator[SlopitSession]:
        """Load multiple sessions from a directory or archive.

        Parameters
        ----------
        path
            Path to directory or archive file.
        pattern
            Glob pattern for file matching.

        Yields
        ------
        SlopitSession
            Session data for each matching file.
        """
        ...

    @classmethod
    @abstractmethod
    def can_load(cls, path: Path) -> bool:
        """Check if this loader can handle the given path.

        Parameters
        ----------
        path
            Path to check.

        Returns
        -------
        bool
            True if this loader can handle the format.
        """
        ...

load abstractmethod

load(path: Path) -> SlopitSession

Load a single session from a file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the data file.	required

Returns:

Type	Description
SlopitSession	The loaded session data.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.
ValueError	If the file format is invalid.

Source code in src/slopit/io/base.py

@abstractmethod
def load(self, path: Path) -> SlopitSession:
    """Load a single session from a file.

    Parameters
    ----------
    path
        Path to the data file.

    Returns
    -------
    SlopitSession
        The loaded session data.

    Raises
    ------
    FileNotFoundError
        If the file does not exist.
    ValueError
        If the file format is invalid.
    """
    ...

load_many abstractmethod

load_many(path: Path, pattern: str = '*') -> Iterator[SlopitSession]

Load multiple sessions from a directory or archive.

Parameters:

Name	Type	Description	Default
path	Path	Path to directory or archive file.	required
pattern	str	Glob pattern for file matching.	'*'

Yields:

Type	Description
SlopitSession	Session data for each matching file.

Source code in src/slopit/io/base.py

@abstractmethod
def load_many(self, path: Path, pattern: str = "*") -> Iterator[SlopitSession]:
    """Load multiple sessions from a directory or archive.

    Parameters
    ----------
    path
        Path to directory or archive file.
    pattern
        Glob pattern for file matching.

    Yields
    ------
    SlopitSession
        Session data for each matching file.
    """
    ...

can_load abstractmethod classmethod

can_load(path: Path) -> bool

Check if this loader can handle the given path.

Parameters:

Name	Type	Description	Default
path	Path	Path to check.	required

Returns:

Type	Description
bool	True if this loader can handle the format.

Source code in src/slopit/io/base.py

@classmethod
@abstractmethod
def can_load(cls, path: Path) -> bool:
    """Check if this loader can handle the given path.

    Parameters
    ----------
    path
        Path to check.

    Returns
    -------
    bool
        True if this loader can handle the format.
    """
    ...

Native Format Loader

NativeLoader

Loader for native slopit JSON format. The native format is a JSON file that directly contains a SlopitSession object.

slopit.io.native.NativeLoader

Bases: BaseLoader

Loader for native slopit JSON format.

The native format is a JSON file that directly contains a SlopitSession object with schemaVersion field.

Examples:

>>> loader = NativeLoader()
>>> session = loader.load(Path("data/session.json"))
>>> print(f"Loaded session {session.session_id}")

Source code in src/slopit/io/native.py

class NativeLoader(BaseLoader):
    """Loader for native slopit JSON format.

    The native format is a JSON file that directly contains a SlopitSession
    object with schemaVersion field.

    Examples
    --------
    >>> loader = NativeLoader()
    >>> session = loader.load(Path("data/session.json"))
    >>> print(f"Loaded session {session.session_id}")
    """

    def load(self, path: Path) -> SlopitSession:
        """Load a native slopit JSON file.

        Parameters
        ----------
        path
            Path to the JSON file.

        Returns
        -------
        SlopitSession
            The loaded session data.
        """
        if not path.exists():
            msg = f"File not found: {path}"
            raise FileNotFoundError(msg)

        content = path.read_text(encoding="utf-8")
        data = json.loads(content)

        return SlopitSession.model_validate(data)

    def load_many(self, path: Path, pattern: str = "*.json") -> Iterator[SlopitSession]:
        """Load multiple native JSON files.

        Parameters
        ----------
        path
            Path to directory containing JSON files.
        pattern
            Glob pattern for file matching.

        Yields
        ------
        SlopitSession
            Session data for each matching file.
        """
        if path.is_file():
            yield self.load(path)
            return

        for file_path in sorted(path.glob(pattern)):
            if file_path.is_file() and self._is_native_format(file_path):
                yield self.load(file_path)

    @classmethod
    def can_load(cls, path: Path) -> bool:
        """Check if path appears to be native slopit format.

        Parameters
        ----------
        path
            Path to check.

        Returns
        -------
        bool
            True if the path appears to be native format.
        """
        if path.is_dir():
            # Check if any JSON files look like native format
            return any(cls._is_native_format(json_file) for json_file in path.glob("*.json"))

        return cls._is_native_format(path)

    @classmethod
    def _is_native_format(cls, path: Path) -> bool:
        """Check if a file is in native slopit format."""
        if path.suffix != ".json":
            return False

        try:
            content = path.read_text(encoding="utf-8")[:1000]
            # Check for both camelCase (JS convention) and snake_case (Python convention)
            has_schema = '"schemaVersion"' in content or '"schema_version"' in content
            has_session = '"sessionId"' in content or '"session_id"' in content
            return has_schema and has_session
        except Exception:
            return False

load

load(path: Path) -> SlopitSession

Load a native slopit JSON file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the JSON file.	required

Returns:

Type	Description
SlopitSession	The loaded session data.

Source code in src/slopit/io/native.py

def load(self, path: Path) -> SlopitSession:
    """Load a native slopit JSON file.

    Parameters
    ----------
    path
        Path to the JSON file.

    Returns
    -------
    SlopitSession
        The loaded session data.
    """
    if not path.exists():
        msg = f"File not found: {path}"
        raise FileNotFoundError(msg)

    content = path.read_text(encoding="utf-8")
    data = json.loads(content)

    return SlopitSession.model_validate(data)

load_many

load_many(path: Path, pattern: str = '*.json') -> Iterator[SlopitSession]

Load multiple native JSON files.

Parameters:

Name	Type	Description	Default
path	Path	Path to directory containing JSON files.	required
pattern	str	Glob pattern for file matching.	'*.json'

Yields:

Type	Description
SlopitSession	Session data for each matching file.

Source code in src/slopit/io/native.py

def load_many(self, path: Path, pattern: str = "*.json") -> Iterator[SlopitSession]:
    """Load multiple native JSON files.

    Parameters
    ----------
    path
        Path to directory containing JSON files.
    pattern
        Glob pattern for file matching.

    Yields
    ------
    SlopitSession
        Session data for each matching file.
    """
    if path.is_file():
        yield self.load(path)
        return

    for file_path in sorted(path.glob(pattern)):
        if file_path.is_file() and self._is_native_format(file_path):
            yield self.load(file_path)

can_load classmethod

can_load(path: Path) -> bool

Check if path appears to be native slopit format.

Parameters:

Name	Type	Description	Default
path	Path	Path to check.	required

Returns:

Type	Description
bool	True if the path appears to be native format.

Source code in src/slopit/io/native.py

@classmethod
def can_load(cls, path: Path) -> bool:
    """Check if path appears to be native slopit format.

    Parameters
    ----------
    path
        Path to check.

    Returns
    -------
    bool
        True if the path appears to be native format.
    """
    if path.is_dir():
        # Check if any JSON files look like native format
        return any(cls._is_native_format(json_file) for json_file in path.glob("*.json"))

    return cls._is_native_format(path)

Native Format Structure

{
  "schemaVersion": "1.0",
  "sessionId": "abc123",
  "participantId": "P001",
  "studyId": "study_001",
  "platform": {
    "name": "jspsych",
    "version": "7.3.0",
    "adapterVersion": "0.1.0"
  },
  "environment": {
    "userAgent": "Mozilla/5.0...",
    "screenResolution": [1920, 1080],
    "viewportSize": [1200, 800],
    "devicePixelRatio": 2.0,
    "timezone": "America/New_York",
    "language": "en-US",
    "touchCapable": false
  },
  "timing": {
    "startTime": 1705000000000,
    "endTime": 1705000600000,
    "duration": 600000
  },
  "trials": [
    {
      "trialId": "trial-0",
      "trialIndex": 0,
      "trialType": "survey-text",
      "startTime": 1705000000000,
      "endTime": 1705000060000,
      "rt": 60000,
      "stimulus": {...},
      "response": {...},
      "behavioral": {...}
    }
  ],
  "globalEvents": {
    "focus": [],
    "errors": []
  }
}

Example

from slopit.io import NativeLoader
from pathlib import Path

loader = NativeLoader()

# Check if file is native format
if NativeLoader.can_load(Path("data/session.json")):
    session = loader.load(Path("data/session.json"))

# Load multiple files
for session in loader.load_many(Path("data/"), pattern="*.json"):
    print(f"Loaded {session.session_id}")

JATOS Loader

JATOSLoader

Loader for data exported from JATOS (Just Another Tool for Online Studies). JATOS exports data as JSON arrays or newline-delimited JSON.

slopit.io.jatos.JATOSLoader

Bases: BaseLoader

Loader for JATOS export format.

JATOS exports data as JSON, either as a single array of trials or as newline-delimited JSON (one trial per line). This loader handles both formats.

Examples:

>>> loader = JATOSLoader()
>>> session = loader.load(Path("jatos_results/study_result_123.txt"))
>>> print(f"Loaded {len(session.trials)} trials")

Source code in src/slopit/io/jatos.py

class JATOSLoader(BaseLoader):
    """Loader for JATOS export format.

    JATOS exports data as JSON, either as a single array of trials
    or as newline-delimited JSON (one trial per line). This loader
    handles both formats.

    Examples
    --------
    >>> loader = JATOSLoader()
    >>> session = loader.load(Path("jatos_results/study_result_123.txt"))
    >>> print(f"Loaded {len(session.trials)} trials")
    """

    def load(self, path: Path) -> SlopitSession:
        """Load a JATOS result file.

        Parameters
        ----------
        path
            Path to the JATOS result file (.txt or .json).

        Returns
        -------
        SlopitSession
            Converted session data.
        """
        if not path.exists():
            msg = f"File not found: {path}"
            raise FileNotFoundError(msg)

        raw_data = self._read_jatos_file(path)
        return self._convert_to_session(raw_data, path.stem)

    def load_result(
        self,
        trials: list[dict[str, JsonValue]],
        result_id: str = "unknown",
    ) -> SlopitSession:
        """Load a session from raw trial data.

        This method is useful for converting JATOS API results directly
        without writing to a file first.

        Parameters
        ----------
        trials
            List of raw trial dictionaries from JATOS.
        result_id
            Optional identifier for the result (for logging/debugging).

        Returns
        -------
        SlopitSession
            Converted session data.

        Examples
        --------
        >>> loader = JATOSLoader()
        >>> trials = [{"trial_type": "survey", "response": "..."}]
        >>> session = loader.load_result(trials, result_id="api-123")
        """
        return self._convert_to_session(trials, result_id)

    def load_many(self, path: Path, pattern: str = "*.txt") -> Iterator[SlopitSession]:
        """Load multiple JATOS result files.

        Parameters
        ----------
        path
            Path to directory containing result files.
        pattern
            Glob pattern for file matching.

        Yields
        ------
        SlopitSession
            Session data for each file.
        """
        if path.is_file():
            yield self.load(path)
            return

        for file_path in sorted(path.glob(pattern)):
            if file_path.is_file():
                try:
                    yield self.load(file_path)
                except (ValueError, json.JSONDecodeError):
                    continue

    @classmethod
    def can_load(cls, path: Path) -> bool:
        """Check if path appears to be JATOS format.

        Parameters
        ----------
        path
            Path to check.

        Returns
        -------
        bool
            True if the path appears to be JATOS format.
        """
        if path.is_dir():
            return any(path.glob("study_result_*.txt")) or any(path.glob("*.txt"))

        if path.suffix not in {".txt", ".json"}:
            return False

        # Check for JATOS markers in content
        try:
            content = path.read_text(encoding="utf-8")[:1000]
            return "trial_type" in content or "jsPsych" in content.lower()
        except Exception:
            return False

    def _read_jatos_file(self, path: Path) -> list[dict[str, JsonValue]]:
        """Read and parse a JATOS result file.

        Handles both array format and newline-delimited JSON.

        Parameters
        ----------
        path
            Path to the file.

        Returns
        -------
        list[dict[str, JsonValue]]
            List of trial data dictionaries.

        Raises
        ------
        json.JSONDecodeError
            If the file does not contain valid JSON.
        """
        content = path.read_text(encoding="utf-8").strip()

        # Try parsing as JSON array first
        if content.startswith("["):
            parsed: list[dict[str, JsonValue]] | dict[str, JsonValue] | JsonValue
            parsed = json.loads(content)
            if isinstance(parsed, list):
                result: list[dict[str, JsonValue]] = []
                for raw_item in parsed:
                    item: JsonValue = raw_item
                    if isinstance(item, dict):
                        # Cast to appropriate type for type checker
                        typed_item: dict[str, JsonValue] = item  # type: ignore[assignment]
                        result.append(typed_item)
                return result
            return []

        # Try newline-delimited JSON
        trials: list[dict[str, JsonValue]] = []
        parsed_any = False
        for line in content.split("\n"):
            line = line.strip()
            if line:
                try:
                    data: JsonValue = json.loads(line)
                    parsed_any = True
                    if isinstance(data, list):
                        for raw_item in data:
                            item: JsonValue = raw_item
                            if isinstance(item, dict):
                                typed_item: dict[str, JsonValue] = item  # type: ignore[assignment]
                                trials.append(typed_item)
                    elif isinstance(data, dict):
                        typed_data: dict[str, JsonValue] = data  # type: ignore[assignment]
                        trials.append(typed_data)
                except json.JSONDecodeError:
                    continue

        if not parsed_any and content:
            # No valid JSON was parsed, raise an error
            msg = "Could not parse any valid JSON from file"
            raise json.JSONDecodeError(msg, content, 0)

        return trials

    def _convert_to_session(
        self,
        trials: list[dict[str, JsonValue]],
        file_id: str,  # noqa: ARG002
    ) -> SlopitSession:
        """Convert raw JATOS trial data to SlopitSession.

        Parameters
        ----------
        trials
            List of raw trial dictionaries.
        file_id
            Identifier derived from filename (reserved for future use).

        Returns
        -------
        SlopitSession
            Converted session.
        """
        # Extract metadata from first trial or use defaults
        first_trial = trials[0] if trials else {}
        last_trial = trials[-1] if trials else {}

        # Determine timing
        if trials:
            time_elapsed = _get_int(first_trial, "time_elapsed", 0)
            rt = _get_int(first_trial, "rt", 0)
            start_time = time_elapsed - rt
            end_time = _get_int(last_trial, "time_elapsed", 0)
        else:
            start_time = end_time = 0

        return SlopitSession(
            schema_version="1.0",
            session_id=str(uuid4()),
            participant_id=self._extract_participant_id(first_trial),
            study_id=self._extract_study_id(first_trial),
            platform=PlatformInfo(
                name="jspsych",
                version=self._detect_jspsych_version(first_trial),
                adapter_version="0.1.0",
            ),
            environment=self._extract_environment(first_trial),
            timing=SessionTiming(
                start_time=start_time,
                end_time=end_time,
                duration=end_time - start_time if end_time > start_time else None,
            ),
            trials=[self._convert_trial(t, i) for i, t in enumerate(trials)],
            global_events=GlobalEvents(),
        )

    def _convert_trial(self, trial: dict[str, JsonValue], index: int) -> SlopitTrial:
        """Convert a single JATOS trial to SlopitTrial.

        Parameters
        ----------
        trial
            Raw trial dictionary.
        index
            Trial index in session.

        Returns
        -------
        SlopitTrial
            Converted trial.
        """
        time_elapsed = _get_int(trial, "time_elapsed", 0)
        rt = _get_int(trial, "rt", 0)

        trial_id_value = trial.get("trial_id")
        trial_id = str(trial_id_value) if trial_id_value is not None else f"trial-{index}"

        trial_type_value = trial.get("trial_type")
        trial_type = str(trial_type_value) if trial_type_value is not None else "unknown"

        converted = SlopitTrial(
            trial_id=trial_id,
            trial_index=index,
            trial_type=trial_type,
            start_time=time_elapsed - rt,
            end_time=time_elapsed,
            rt=rt,
        )

        # Extract response
        if "response" in trial:
            response = trial["response"]
            converted.response = ResponseInfo(
                type="text" if isinstance(response, str) else "other",
                value=response,
                character_count=len(response) if isinstance(response, str) else None,
                word_count=len(response.split()) if isinstance(response, str) else None,
            )

        # Extract slopit data
        sd_data = trial.get("slopit")
        if isinstance(sd_data, dict):
            behavioral_data = sd_data.get("behavioral")
            if isinstance(behavioral_data, dict):
                converted.behavioral = BehavioralData.model_validate(behavioral_data)
            flags_data = sd_data.get("flags")
            if isinstance(flags_data, list):
                converted.capture_flags = flags_data  # type: ignore[assignment]

        # Store platform data
        platform_data = {k: v for k, v in trial.items() if k != "slopit"}
        converted.platform_data = platform_data

        return converted

    def _extract_participant_id(self, trial: dict[str, JsonValue]) -> str | None:
        """Extract participant ID from trial data."""
        for key in ["PROLIFIC_PID", "workerId", "participant_id", "subject"]:
            if key in trial:
                value = trial[key]
                if value is not None:
                    return str(value)
        return None

    def _extract_study_id(self, trial: dict[str, JsonValue]) -> str | None:
        """Extract study ID from trial data."""
        for key in ["STUDY_ID", "study_id", "experiment_id"]:
            if key in trial:
                value = trial[key]
                if value is not None:
                    return str(value)
        return None

    def _detect_jspsych_version(self, trial: dict[str, JsonValue]) -> str | None:
        """Attempt to detect jsPsych version."""
        version = trial.get("jspsych_version")
        if version is not None:
            return str(version)
        return None

    def _extract_environment(self, trial: dict[str, JsonValue]) -> EnvironmentInfo:
        """Extract environment info from trial data."""
        return EnvironmentInfo(
            user_agent=_get_str(trial, "user_agent", "unknown"),
            screen_resolution=(
                _get_int(trial, "screen_width", 0),
                _get_int(trial, "screen_height", 0),
            ),
            viewport_size=(
                _get_int(trial, "viewport_width", 0),
                _get_int(trial, "viewport_height", 0),
            ),
            device_pixel_ratio=_get_float(trial, "device_pixel_ratio", 1.0),
            timezone=_get_str(trial, "timezone", "unknown"),
            language=_get_str(trial, "language", "unknown"),
            touch_capable=_get_bool(trial, "touch_capable", False),
        )

load

load(path: Path) -> SlopitSession

Load a JATOS result file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the JATOS result file (.txt or .json).	required

Returns:

Type	Description
SlopitSession	Converted session data.

Source code in src/slopit/io/jatos.py

def load(self, path: Path) -> SlopitSession:
    """Load a JATOS result file.

    Parameters
    ----------
    path
        Path to the JATOS result file (.txt or .json).

    Returns
    -------
    SlopitSession
        Converted session data.
    """
    if not path.exists():
        msg = f"File not found: {path}"
        raise FileNotFoundError(msg)

    raw_data = self._read_jatos_file(path)
    return self._convert_to_session(raw_data, path.stem)

load_result

load_result(trials: list[dict[str, JsonValue]], result_id: str = 'unknown') -> SlopitSession

Load a session from raw trial data.

This method is useful for converting JATOS API results directly without writing to a file first.

Parameters:

Name	Type	Description	Default
trials	list[dict[str, JsonValue]]	List of raw trial dictionaries from JATOS.	required
result_id	str	Optional identifier for the result (for logging/debugging).	'unknown'

Returns:

Type	Description
SlopitSession	Converted session data.

Examples:

>>> loader = JATOSLoader()
>>> trials = [{"trial_type": "survey", "response": "..."}]
>>> session = loader.load_result(trials, result_id="api-123")

Source code in src/slopit/io/jatos.py

def load_result(
    self,
    trials: list[dict[str, JsonValue]],
    result_id: str = "unknown",
) -> SlopitSession:
    """Load a session from raw trial data.

    This method is useful for converting JATOS API results directly
    without writing to a file first.

    Parameters
    ----------
    trials
        List of raw trial dictionaries from JATOS.
    result_id
        Optional identifier for the result (for logging/debugging).

    Returns
    -------
    SlopitSession
        Converted session data.

    Examples
    --------
    >>> loader = JATOSLoader()
    >>> trials = [{"trial_type": "survey", "response": "..."}]
    >>> session = loader.load_result(trials, result_id="api-123")
    """
    return self._convert_to_session(trials, result_id)

load_many

load_many(path: Path, pattern: str = '*.txt') -> Iterator[SlopitSession]

Load multiple JATOS result files.

Parameters:

Name	Type	Description	Default
path	Path	Path to directory containing result files.	required
pattern	str	Glob pattern for file matching.	'*.txt'

Yields:

Type	Description
SlopitSession	Session data for each file.

Source code in src/slopit/io/jatos.py

def load_many(self, path: Path, pattern: str = "*.txt") -> Iterator[SlopitSession]:
    """Load multiple JATOS result files.

    Parameters
    ----------
    path
        Path to directory containing result files.
    pattern
        Glob pattern for file matching.

    Yields
    ------
    SlopitSession
        Session data for each file.
    """
    if path.is_file():
        yield self.load(path)
        return

    for file_path in sorted(path.glob(pattern)):
        if file_path.is_file():
            try:
                yield self.load(file_path)
            except (ValueError, json.JSONDecodeError):
                continue

can_load classmethod

can_load(path: Path) -> bool

Check if path appears to be JATOS format.

Parameters:

Name	Type	Description	Default
path	Path	Path to check.	required

Returns:

Type	Description
bool	True if the path appears to be JATOS format.

Source code in src/slopit/io/jatos.py

@classmethod
def can_load(cls, path: Path) -> bool:
    """Check if path appears to be JATOS format.

    Parameters
    ----------
    path
        Path to check.

    Returns
    -------
    bool
        True if the path appears to be JATOS format.
    """
    if path.is_dir():
        return any(path.glob("study_result_*.txt")) or any(path.glob("*.txt"))

    if path.suffix not in {".txt", ".json"}:
        return False

    # Check for JATOS markers in content
    try:
        content = path.read_text(encoding="utf-8")[:1000]
        return "trial_type" in content or "jsPsych" in content.lower()
    except Exception:
        return False

JATOS Format

JATOS exports come in two formats:

Array format (single JSON array):

[
  {"trial_type": "html-keyboard-response", "rt": 1234, ...},
  {"trial_type": "survey-text", "response": "...", "slopit": {...}, ...}
]

Newline-delimited format (one trial per line):

{"trial_type": "html-keyboard-response", "rt": 1234, ...}
{"trial_type": "survey-text", "response": "...", "slopit": {...}, ...}

Extracted Fields

The JATOS loader extracts:

• participant_id: From PROLIFIC_PID, workerId, participant_id, or subject
• study_id: From STUDY_ID, study_id, or experiment_id
• environment: From standard jsPsych fields (user_agent, screen_width, etc.)
• behavioral data: From the slopit field added by slopit adapters

Example

from slopit.io import JATOSLoader
from pathlib import Path

loader = JATOSLoader()

# Load JATOS result file
session = loader.load(Path("jatos_results/study_result_123.txt"))
print(f"Loaded {len(session.trials)} trials")

# Load all results from a directory
for session in loader.load_many(Path("jatos_results/")):
    print(f"Session: {session.session_id}")
    print(f"  Participant: {session.participant_id}")
    print(f"  Trials: {len(session.trials)}")

Writing Custom Loaders

To support a new data format, subclass BaseLoader:

from pathlib import Path
from collections.abc import Iterator
from slopit.io.base import BaseLoader
from slopit.schemas import SlopitSession

class MyCustomLoader(BaseLoader):
    """Loader for my custom format."""

    def load(self, path: Path) -> SlopitSession:
        """Load a single session."""
        # Parse your format
        raw_data = self._parse_file(path)

        # Convert to SlopitSession
        return SlopitSession(
            schema_version="1.0",
            session_id=raw_data["id"],
            # ... map other fields
        )

    def load_many(self, path: Path, pattern: str = "*") -> Iterator[SlopitSession]:
        """Load multiple sessions."""
        if path.is_file():
            yield self.load(path)
            return

        for file_path in sorted(path.glob(pattern)):
            if self._is_my_format(file_path):
                yield self.load(file_path)

    @classmethod
    def can_load(cls, path: Path) -> bool:
        """Check if this loader can handle the path."""
        if path.is_dir():
            return any(cls._is_my_format(f) for f in path.glob("*"))
        return cls._is_my_format(path)

    @classmethod
    def _is_my_format(cls, path: Path) -> bool:
        """Check if file is in my custom format."""
        # Implement format detection
        return path.suffix == ".myformat"

    def _parse_file(self, path: Path) -> dict:
        """Parse file contents."""
        # Implement parsing logic
        pass

Registering Custom Loaders

Currently, custom loaders must be used directly:

loader = MyCustomLoader()
session = loader.load(Path("data/file.myformat"))

Future versions will support registering loaders for automatic format detection.