"""
Reading and writing data.
- Eyelink EDF files: :func:`read_eyelink` (usually used through
:meth:`~pypillometry.EyeData.from_eyelink`)
- saved objects: :func:`write_file` and :func:`read_file` (pickle or
HDF5 files; usually used through :meth:`~pypillometry.EyeData.write_file`
and :meth:`~pypillometry.EyeData.from_file`)
- files from URLs (HTTP(S), SSH, OSF) and the download cache:
:func:`fetch_url`, :func:`local_copy`, :func:`download`,
:func:`osf_authenticate`
- files from public figshare items: :func:`get_figshare_files`,
:func:`fetch_figshare_file`
See the user guide (Loading and saving data) for an overview.
"""
import gzip
import pickle
import os
from contextlib import contextmanager
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING: # imported where used (slow)
import requests
from loguru import logger
from .convenience import is_url, suppress_all_output, requires_package
from .logging import logging_get_level, _show_progress
from . import state
from . import hdf5
try:
from compression import zstd # Python >= 3.14
except ImportError:
from backports import zstd
#: zstd compression level used by :func:`write_file`
ZSTD_LEVEL = 3
_ZSTD_MAGIC = b"\x28\xb5\x2f\xfd"
_GZIP_MAGIC = b"\x1f\x8b"
_HDF5_EXTENSIONS = (".h5", ".hdf5")
__all__ = [
"osf_authenticate", "download", "fetch_url", "local_copy", "write_file",
"read_file", "read_eyelink", "get_figshare_files", "fetch_figshare_file",
]
[docs]
def osf_authenticate(
access_token: str,
validate: bool = True,
session: Optional["requests.Session"] = None,
timeout: float = 10.0,
) -> "requests.Session":
"""
Authenticate against the OSF API using a personal access token.
Parameters
----------
access_token : str
Personal access token generated on OSF.
validate : bool, optional
If True (default), verify the token by calling ``/v2/users/me/``.
session : requests.Session, optional
Existing session to configure. If None, a new session is created.
timeout : float, optional
Timeout (in seconds) for the validation request when ``validate`` is True.
Returns
-------
requests.Session
Session configured with the OSF bearer token.
Raises
------
ValueError
If no access token is provided or validation fails.
"""
import requests
if not access_token or not isinstance(access_token, str):
raise ValueError("A non-empty OSF access token string is required.")
token = access_token.strip()
if not token:
raise ValueError("Access token cannot be empty or whitespace.")
session = session or requests.Session()
session.headers.update(
{
"Authorization": f"Bearer {token}",
"Accept": "application/json",
}
)
if validate:
try:
response = session.get("https://api.osf.io/v2/users/me/", timeout=timeout)
response.raise_for_status()
except requests.exceptions.HTTPError as exc:
status = exc.response.status_code if exc.response is not None else "unknown"
raise ValueError(f"OSF token validation failed (status {status}).") from exc
except requests.exceptions.RequestException as exc:
raise ValueError("OSF token validation request failed.") from exc
else:
logger.info("OSF token validation succeeded.")
return session
def _osf_client_url_to_download_url(url: str, session: Optional["requests.Session"] = None) -> str:
"""
Convert an OSF client URL to a download URL.
Parameters
----------
url : str
OSF client URL
session : requests.Session, optional
requests session to use for the download. If None, a new session is created.
Returns
-------
str
Download URL
"""
import requests
# is it an OSF client URL?
if not url.startswith("https://osf.io/"):
logger.debug(f"URL is not an OSF client URL: {url!r}")
return url
# Check if it's already a download URL
if "/download/" in url:
logger.debug(f"URL is already a download URL: {url!r}")
return url
# get the project ID
from urllib.parse import urlsplit
parts = [segment for segment in urlsplit(url).path.split('/') if segment]
if not parts:
logger.error(f"No OSF identifier found in {url!r}")
return url
file_id = parts[0]
logger.debug(f"File ID: {file_id}")
# make a get request to https://api.osf.io/v2/files/{file_id}/
if session is not None:
response = session.get(f"https://api.osf.io/v2/files/{file_id}/")
else:
response = requests.get(f"https://api.osf.io/v2/files/{file_id}/")
download_url = response.json()["data"]["links"]["move"].replace("/move", "/download")
logger.debug(f"Download URL: {download_url}")
return download_url
[docs]
def download(url: str, fname: str = None, chunk_size: int = 1024, session: Optional["requests.Session"] = None) -> str:
"""Download a file from a URL to a local file.
See https://gist.github.com/yanqd0/c13ed29e29432e3cf3e7c38467f42f51.
Parameters
----------
url : str
URL of the file to download.
fname : str, optional
Local filename to save the file to. If None, create a temporary file.
chunk_size : int, optional
Size of the chunks to download the file in. Default is 1024.
session : requests.Session, optional
requests session to use for the download. If None, a new session is created.
Returns
-------
str
Local filename of the downloaded file.
"""
import requests
from tqdm import tqdm
import tempfile
# Create temporary file if fname is None
if fname is None:
# Extract file extension from URL if possible
from urllib.parse import urlparse
parsed_url = urlparse(url)
path = parsed_url.path
if path and "." in os.path.basename(path):
suffix = os.path.splitext(path)[1]
else:
suffix = ""
# Create temporary file
fd, fname = tempfile.mkstemp(suffix=suffix)
os.close(fd) # Close the file descriptor, we'll open it again below
# convert the URL to a download URL if it is an OSF client URL
url = _osf_client_url_to_download_url(url, session=session)
own_session = session is None
session = session or requests.Session()
try:
with session.get(url, stream=True) as resp:
resp.raise_for_status()
# unknown size (no content-length): progress bar without total
total = int(resp.headers.get("content-length", 0)) or None
# fetch_url() downloads to "<name>.part"
name = os.path.basename(fname)
if name.endswith(".part"):
name = name[:-len(".part")]
with open(fname, "wb") as file, tqdm(
desc=name,
total=total,
unit="iB",
unit_scale=True,
unit_divisor=1024,
disable=not _show_progress(),
) as bar:
for data in resp.iter_content(chunk_size=chunk_size):
size = file.write(data)
bar.update(size)
finally:
if own_session:
session.close()
return fname
def is_ssh_url(source) -> bool:
"""Check if `source` is an SSH URL (``ssh://[user@]host[:port]/path``).
Parameters
----------
source : str
string to check
Returns
-------
bool
True if `source` starts with ``ssh://``
"""
return isinstance(source, str) and source.startswith("ssh://")
def _parse_ssh_url(url: str):
"""Split an SSH URL into (host, port, remote path).
`host` may contain a user (``user@host``). The path is absolute
(``ssh://host/data/f.edf`` -> ``/data/f.edf``) unless it starts with
``~/``; then it is relative to the home directory on the server
(``ssh://host/~/data/f.edf`` -> ``data/f.edf``).
"""
from urllib.parse import urlsplit, unquote
parts = urlsplit(url)
if parts.scheme != "ssh" or not parts.hostname:
raise ValueError(f"Not a valid SSH URL: '{url}' (expected ssh://host/path)")
host = parts.hostname
if parts.username:
host = f"{parts.username}@{host}"
path = unquote(parts.path)
if path in ("", "/"):
raise ValueError(f"SSH URL '{url}' has no path")
if path.startswith("/~/"):
path = path[3:]
elif path == "/~":
raise ValueError(f"SSH URL '{url}' has no path")
return host, parts.port, path
def _cache_path(url: str, cache_dir: str, layout: str = "flat") -> str:
"""Local path in `cache_dir` for a URL.
With ``layout="flat"``, the file (or folder) name only:
``cache_dir/<name>``. With ``layout="mirror"``, the full remote path:
``cache_dir/<host>/<path>``; for SSH URLs with a path relative to the
home directory, ``~`` is inserted after the host.
"""
from urllib.parse import urlsplit, unquote
if layout not in ("flat", "mirror"):
raise ValueError(f"cache_layout must be 'flat' or 'mirror', not '{layout}'")
if is_ssh_url(url):
host, port, path = _parse_ssh_url(url)
rel = path.strip("/").split("/")
if not path.startswith("/"):
rel = ["~"] + rel # relative to home on the server
else:
parts = urlsplit(url)
host, port = parts.hostname, parts.port
rel = [p for p in unquote(parts.path).split("/") if p] or ["index"]
if layout == "flat":
return os.path.join(cache_dir, rel[-1])
host_dir = host if port is None else f"{host}_{port}"
return os.path.join(cache_dir, host_dir, *rel)
def _ssh_remote_size(host: str, port, path: str) -> Optional[int]:
"""Size in bytes of a file (or folder, if `path` ends with ``/``) on an SSH server.
Uses ``wc -c`` for files and ``du -sk`` for folders (approximate).
Returns None if the size cannot be determined.
"""
import shlex
import shutil
import subprocess
if shutil.which("ssh") is None:
return None
is_dir = path.endswith("/")
qpath = shlex.quote(path.rstrip("/") if is_dir else path)
remote = f"du -sk {qpath}" if is_dir else f"wc -c < {qpath}"
cmd = ["ssh", "-o", "BatchMode=yes"]
if port is not None:
cmd += ["-p", str(port)]
cmd += [host, remote]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
size = int(result.stdout.split()[0]) if result.returncode == 0 else None
except (OSError, subprocess.TimeoutExpired, ValueError, IndexError):
size = None
if size is None:
logger.debug(f"Could not get the size of '{host}:{path}'")
return None
return size * 1024 if is_dir else size
def _local_size(path: str) -> int:
"""Size in bytes of a local file or folder (0 if it does not exist)."""
if os.path.isfile(path):
return os.path.getsize(path)
total = 0
for root, _, fnames in os.walk(path):
for fname in fnames:
try:
total += os.path.getsize(os.path.join(root, fname))
except OSError: # file removed while walking
pass
return total
def _scp(url: str, target: str):
"""Copy the file (or folder, if `url` ends with ``/``) at an SSH URL to `target`.
Shows a progress bar that follows the size of `target` while ``scp``
is running.
"""
from tqdm import tqdm
import subprocess
import shutil
import threading
if shutil.which("scp") is None:
raise RuntimeError("Downloading from SSH URLs requires the 'scp' command")
host, port, path = _parse_ssh_url(url)
is_dir = path.endswith("/")
cmd = ["scp", "-q", "-o", "BatchMode=yes"]
if port is not None:
cmd += ["-P", str(port)]
if is_dir:
cmd.append("-r")
cmd += [f"{host}:{path.rstrip('/') if is_dir else path}", target]
total = _ssh_remote_size(host, port, path)
# run scp in a thread and follow the size of the local copy
out = {}
def run():
try:
out["result"] = subprocess.run(cmd, capture_output=True, text=True)
except BaseException as e:
out["error"] = e
worker = threading.Thread(target=run, daemon=True)
worker.start()
with tqdm(desc=os.path.basename(path.rstrip("/")), total=total,
unit="iB", unit_scale=True, unit_divisor=1024,
disable=not _show_progress()) as bar:
while worker.is_alive():
worker.join(0.2)
bar.update(_local_size(target) - bar.n)
# final size (also if scp finished before the first update)
bar.update(_local_size(target) - bar.n)
if "error" in out:
raise out["error"]
result = out["result"]
if result.returncode != 0:
msg = result.stderr.strip()
raise RuntimeError(
f"Could not download '{url}' via SSH (scp exit code "
f"{result.returncode}): {msg}\n"
"Note: only key-based SSH login is supported (no passwords)."
)
[docs]
def fetch_url(
url: str,
cache_file: Optional[str] = None,
cache_dir: Optional[str] = None,
force_download: bool = False,
session: Optional["requests.Session"] = None,
cache_layout: str = "flat") -> str:
"""Download an HTTP(S) or SSH URL to a local file, with caching.
HTTP(S) URLs are downloaded with :func:`download` (OSF client URLs
such as ``https://osf.io/<id>`` are converted to download URLs).
SSH URLs have the form ``ssh://[user@]host[:port]/path`` and are
downloaded with the system's ``scp``, which uses ``~/.ssh/config``
(host aliases, users, keys). Only key-based login is supported: password
prompts are disabled (``BatchMode=yes``). The path is absolute on the
server; use ``ssh://host/~/path`` for a path relative to the home
directory. An SSH URL ending in ``/`` downloads the whole folder.
Parameters
----------
url : str
HTTP(S) or SSH URL
cache_file : str, optional
local filename to store the download in
cache_dir : str, optional
directory to store the download in (used if `cache_file` is None);
see `cache_layout`. If both are None, a new temporary directory is
used. The location is logged.
force_download : bool, optional
if False (default), an existing local copy is reused; if True, the
file is downloaded again
session : requests.Session, optional
requests session for HTTP(S) downloads (e.g., from
:func:`osf_authenticate` for private OSF projects)
cache_layout : str, optional
where in `cache_dir` the download is stored: ``"flat"`` (default)
uses the file name only (``cache_dir/<file>``), ``"mirror"`` the
full remote path (``cache_dir/<host>/<path>``). With ``"flat"``,
files with the same name from different URLs share one local copy,
so the first one would be reused; use ``"mirror"`` (or
`force_download`) in that case.
Returns
-------
str
local path of the downloaded file (or folder)
"""
import shutil
import tempfile
if not is_url(url): # also true for SSH URLs
raise ValueError(f"Not a URL: '{url}'")
if cache_file is None:
if cache_dir is None:
cache_dir = tempfile.mkdtemp(prefix="pypillometry_")
cache_file = _cache_path(url, cache_dir, cache_layout)
if os.path.exists(cache_file) and not force_download:
logger.info(f"Using cached copy of '{url}': {cache_file}")
return cache_file
parent = os.path.dirname(cache_file)
if parent:
os.makedirs(parent, exist_ok=True)
# download to a temporary name, so that an interrupted download
# is not used as a cached copy later
part = cache_file + ".part"
if os.path.isdir(part):
shutil.rmtree(part)
elif os.path.exists(part):
os.remove(part)
logger.info(f"Downloading '{url}' to {cache_file}")
if is_ssh_url(url):
_scp(url, part)
else:
download(url, part, session=session)
if os.path.isdir(cache_file):
shutil.rmtree(cache_file)
os.replace(part, cache_file)
return cache_file
_FIGSHARE_API = "https://api.figshare.com/v2"
def _figshare_article(article) -> tuple:
"""(item ID, version or None) from an item ID, a figshare URL or a DOI."""
import re
s = str(article).strip()
if s.isdigit():
return s, None
if "figshare" in s:
# DOI: 10.6084/m9.figshare.<id>[.v<version>]
m = re.search(r"figshare\.(\d+)(?:\.v(\d+))?/?$", s)
# URL: https://figshare.com/articles/<type>/<title>/<id>[/<version>]
m = m or re.search(r"/(\d+)(?:/(\d+))?/?(?:[?#].*)?$", s)
if m:
return m.group(1), int(m.group(2)) if m.group(2) else None
raise ValueError(f"Not a figshare item ID, URL or DOI: '{article}'")
[docs]
def get_figshare_files(article, version: Optional[int] = None) -> dict:
"""List the files of a public figshare item.
Parameters
----------
article : int or str
figshare item ID (e.g., ``28776617``), the item's URL
(``https://figshare.com/articles/dataset/<title>/28776617``) or its
DOI (``10.6084/m9.figshare.28776617``)
version : int, optional
version of the item; default the latest one (or the version in the
URL or DOI, if it contains one)
Returns
-------
dict
file name -> dict with the file's ``id``, ``download_url``, ``size``
(bytes) and ``md5`` checksum
Examples
--------
>>> files = get_figshare_files(28776617) # doctest: +SKIP
>>> files["sub-001_ses-1.pypil"]["size"] # doctest: +SKIP
30879440
"""
import requests
article_id, url_version = _figshare_article(article)
if version is None:
version = url_version
url = f"{_FIGSHARE_API}/articles/{article_id}"
if version is not None:
url += f"/versions/{version}"
url += "/files"
files = {}
page, page_size = 1, 1000
while True:
resp = requests.get(url, params={"page": page, "page_size": page_size},
timeout=30)
if resp.status_code == 404:
which = f"item {article_id}" + (f", version {version}," if version else "")
raise ValueError(f"figshare {which} not found (only public items "
"are supported)")
resp.raise_for_status()
batch = resp.json()
for f in batch:
files[f["name"]] = {
"id": f["id"],
"download_url": f["download_url"],
"size": f.get("size", 0),
"md5": f.get("computed_md5") or f.get("supplied_md5") or "",
}
if len(batch) < page_size:
return files
page += 1
def _md5(fname: str) -> str:
import hashlib
h = hashlib.md5()
with open(fname, "rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
[docs]
def fetch_figshare_file(
article,
name: str,
version: Optional[int] = None,
cache_dir: Optional[str] = None,
force_download: bool = False,
cache_layout: str = "flat") -> str:
"""Download a file from a public figshare item, with caching.
The download is checked against the MD5 checksum that figshare provides.
Parameters
----------
article : int or str
figshare item ID, URL or DOI (see :func:`get_figshare_files`)
name : str
name of the file in the item (see :func:`get_figshare_files`)
version : int, optional
version of the item; default the latest one (or the version in the
URL or DOI, if it contains one)
cache_dir : str, optional
directory to store the download in; if None, a new temporary
directory is used
force_download : bool, optional
if False (default), an existing local copy is reused (without
contacting figshare); if True, the file is downloaded again
cache_layout : str, optional
``"flat"`` (default) stores the file as ``cache_dir/<name>``,
``"mirror"`` as ``cache_dir/figshare.com/<id>/<version>/<name>``
(``latest`` if no version is given), so that files with the same
name from different items do not share a local copy
Returns
-------
str
local path of the downloaded file
Examples
--------
>>> fname = fetch_figshare_file(28776617, "sub-001_ses-1.pypil",
... cache_dir="data") # doctest: +SKIP
>>> d = EyeData.from_file(fname) # doctest: +SKIP
"""
import tempfile
article_id, url_version = _figshare_article(article)
if version is None:
version = url_version
if os.path.basename(name) != name or name in ("", ".", ".."):
raise ValueError(f"Not a file name: '{name}'")
if cache_dir is None:
cache_dir = tempfile.mkdtemp(prefix="pypillometry_")
if cache_layout == "flat":
cache_file = os.path.join(cache_dir, name)
elif cache_layout == "mirror":
cache_file = os.path.join(cache_dir, "figshare.com", article_id,
f"v{version}" if version else "latest", name)
else:
raise ValueError(f"cache_layout must be 'flat' or 'mirror', not '{cache_layout}'")
if os.path.exists(cache_file) and not force_download:
logger.info(f"Using cached copy of '{name}' (figshare {article_id}): {cache_file}")
return cache_file
files = get_figshare_files(article_id, version)
if name not in files:
raise ValueError(f"File '{name}' not found in figshare item {article_id}; "
f"available: {', '.join(files)}")
info = files[name]
fetch_url(info["download_url"], cache_file=cache_file, force_download=True)
if info["md5"] and _md5(cache_file) != info["md5"]:
os.remove(cache_file)
raise ValueError(f"Download of '{name}' (figshare {article_id}) is "
"corrupted (MD5 checksum does not match); try again")
return cache_file
[docs]
@contextmanager
def local_copy(
source: str,
cache_file: Optional[str] = None,
cache_dir: Optional[str] = None,
autodelete: Optional[bool] = None,
force_download: bool = False,
session: Optional["requests.Session"] = None,
cache_layout: str = "flat"):
"""Context manager that gives a local filename for `source`.
HTTP(S) and SSH URLs are downloaded with :func:`fetch_url`; local
paths are passed through unchanged (and never deleted).
Parameters
----------
source : str
local path, HTTP(S) URL or SSH URL
cache_file, cache_dir, force_download, session, cache_layout
see :func:`fetch_url`
autodelete : bool or None, optional
if True, the downloaded copy (also in `cache_file`/`cache_dir`) is
deleted when the context is left, i.e. after the file was read; a
temporary directory is removed completely. If None (default), only
a temporary directory (no `cache_file`/`cache_dir`) is removed.
Yields
------
str
local filename
"""
import shutil
import tempfile
if not is_url(source): # also true for SSH URLs
yield source
return
tmpdir = None
if cache_file is None and cache_dir is None:
tmpdir = tempfile.mkdtemp(prefix="pypillometry_")
cache_dir = tmpdir
if autodelete is None:
autodelete = tmpdir is not None
fname = fetch_url(source, cache_file=cache_file, cache_dir=cache_dir,
force_download=force_download, session=session,
cache_layout=cache_layout)
try:
yield fname
finally:
if autodelete:
target = tmpdir or fname
logger.info(f"Deleting downloaded copy: {target}")
if os.path.isdir(target):
shutil.rmtree(target, ignore_errors=True)
elif os.path.exists(target):
os.remove(target)
def _is_hdf5_name(fname) -> bool:
"""True if `fname` has an HDF5 extension (``.h5`` or ``.hdf5``)."""
return str(fname).lower().endswith(_HDF5_EXTENSIONS)
[docs]
def write_file(obj, fname, compress: bool = True):
"""
Store any Python object in a file using :mod:`pickle` (or HDF5).
By default, the file is compressed with zstd (about 13 times smaller
for eyedata objects). :func:`read_file` detects the compression
automatically.
If `fname` ends with ``.h5`` or ``.hdf5``, the object is stored in an
HDF5 file instead (see :mod:`pypillometry.hdf5`); this works for
eyedata and segmented objects only.
Parameters
----------
obj : object
object to save
fname : str
filename to save to
compress : bool, optional
if True (default), compress the file with zstd (HDF5: larger arrays
with gzip); if False, write an uncompressed file
Raises
------
TypeError
if `fname` is an HDF5 file and `obj` is not an eyedata or segmented
object
"""
if _is_hdf5_name(fname):
hdf5.write_state(state.to_state(obj), fname, compress=compress)
return
if compress:
f = zstd.open(fname, "wb", level=ZSTD_LEVEL)
else:
f = open(fname, "wb")
with f:
pickle.dump(obj, f, protocol=pickle.HIGHEST_PROTOCOL)
def _open_maybe_compressed(fname):
"""Open `fname` for reading, decompressing zstd or gzip files."""
with open(fname, "rb") as f:
magic = f.read(4)
if magic.startswith(_ZSTD_MAGIC):
return zstd.open(fname, "rb")
if magic.startswith(_GZIP_MAGIC):
return gzip.open(fname, "rb")
return open(fname, "rb")
[docs]
def read_file(
fname: str,
cache_file: Optional[str] = None,
force_download: bool = False,
session: Optional["requests.Session"] = None,
cache_dir: Optional[str] = None,
autodelete: Optional[bool] = None,
cache_layout: str = "flat"):
"""
Read a Python object from a file using :mod:`pickle` (or HDF5).
Plain, zstd- and gzip-compressed pickle files and HDF5 files written
by :func:`write_file` are supported (the format is detected from the
file content). Objects saved with an older version of pypillometry are
upgraded to the current version.
URLs are downloaded with :func:`~pypillometry.io.fetch_url`, in the
same way as in :func:`~pypillometry.io.read_eyelink`.
Parameters
----------
fname : str
filename, HTTP(S) URL or SSH URL (``ssh://server/folder/file.pd``)
cache_file : str, optional
local filename to store a downloaded file in
cache_dir : str, optional
directory to store a downloaded file in (as ``cache_dir/<file>``,
see `cache_layout`), used if `cache_file` is None. If both are
None, a temporary directory is used.
force_download : bool, optional
if True, download again even if a local copy exists in
`cache_file`/`cache_dir`
session : requests.Session, optional
requests session for HTTP(S) downloads (e.g., from
:func:`~pypillometry.io.osf_authenticate` for private projects)
autodelete : bool or None, optional
if True, the downloaded copy is deleted after it was read (also in
`cache_file`/`cache_dir`). If None (default), only a temporary
download (no `cache_file`/`cache_dir`) is deleted. Local files are
never deleted.
cache_layout : str, optional
``"flat"`` (default) or ``"mirror"`` (``cache_dir/<host>/<path>``),
see :func:`~pypillometry.io.fetch_url`
Returns
-------
object
loaded object
Warnings
--------
Pickle files can run arbitrary code when they are read. Only read pickle
files from sources you trust; HDF5 files do not run code.
"""
with local_copy(fname, cache_file=cache_file, cache_dir=cache_dir,
autodelete=autodelete, force_download=force_download,
session=session, cache_layout=cache_layout) as local:
if hdf5.is_hdf5(local):
return state.from_state(hdf5.read_state(local))
with _open_maybe_compressed(local) as f:
# maps classes that moved to their new location
return state._Unpickler(f).load()
[docs]
@requires_package("eyelinkio")
def read_eyelink(
source: str,
cache_file: Optional[str] = None,
force_download: bool = False,
session: Optional["requests.Session"] = None,
cache_dir: Optional[str] = None,
autodelete: Optional[bool] = None,
cache_layout: str = "flat"):
"""
Read an Eyelink file/URL and return the object returned by the "eyelinkio" package.
Requires the optional package `eyelinkio`. Its console output is only
shown when the log level is "DEBUG". URLs are downloaded with
:func:`~pypillometry.io.fetch_url`.
Parameters
----------
source : str
filename of the Eyelink EDF file, HTTP(S) URL (OSF client URLs such
as ``https://osf.io/<id>`` are converted to download URLs) or SSH URL
(``ssh://server/folder/file.edf``)
cache_file : str, optional
local filename to store a downloaded file in
cache_dir : str, optional
directory to store a downloaded file in (as ``cache_dir/<file>``,
see `cache_layout`), used if `cache_file` is None. If both are
None, a temporary directory is used.
force_download : bool, optional
if True, download again even if a local copy exists in
`cache_file`/`cache_dir`
session : requests.Session, optional
requests session for HTTP(S) downloads (e.g., from
:func:`~pypillometry.io.osf_authenticate` for private projects)
autodelete : bool or None, optional
if True, the downloaded copy is deleted after it was read (also in
`cache_file`/`cache_dir`). If None (default), only a temporary
download (no `cache_file`/`cache_dir`) is deleted. Local files are
never deleted.
cache_layout : str, optional
``"flat"`` (default) or ``"mirror"`` (``cache_dir/<host>/<path>``),
see :func:`~pypillometry.io.fetch_url`
Returns
-------
eyelinkio.EDF
object returned by :func:`eyelinkio.read_edf`
Notes
-----
Reading EDF files is slow (about a minute for a 90 MB file, mostly in
`eyelinkio`). Convert them once and work with the saved files, which
are read in well under a second: with the command-line program
``edf2pypil`` or with :meth:`~pypillometry.EyeData.write_file`.
See Also
--------
pypillometry.EyeData.from_eyelink : create an eyedata object from an EDF file
"""
with local_copy(source, cache_file=cache_file, cache_dir=cache_dir,
autodelete=autodelete, force_download=force_download,
session=session, cache_layout=cache_layout) as fname:
return _read_edf(fname)
def _read_edf(fname: str):
"""Read a local EDF file with eyelinkio (its output only shown at DEBUG level)."""
import eyelinkio
# Check if DEBUG logging is enabled
current_level = logging_get_level()
show_eyelinkio_output = (current_level == "DEBUG")
if show_eyelinkio_output:
logger.debug(f"Loading EDF file: {fname}")
logger.debug("eyelinkio output will be displayed below:")
with _c_time_locale():
edf = eyelinkio.read_edf(fname)
else:
logger.info(f"Loading EDF file: {fname} (current log level: {current_level}, set to DEBUG to see eyelinkio output)")
with suppress_all_output(), _c_time_locale():
edf = eyelinkio.read_edf(fname)
return edf
@contextmanager
def _c_time_locale():
"""Use the "C" locale for times while reading an EDF file.
eyelinkio parses the English date in the file header ("Fri Feb 14
...") with ``strptime``, which fails if a program activated a locale
with other names of days and months (e.g., Norwegian).
"""
import locale
old = locale.setlocale(locale.LC_TIME)
try:
locale.setlocale(locale.LC_TIME, "C")
yield
finally:
locale.setlocale(locale.LC_TIME, old)
