#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Part of the PsychoPy library
# Copyright (C) 2002-2018 Jonathan Peirce (C) 2019-2024 Open Science Tools Ltd.
# Distributed under the terms of the GNU General Public License (GPL).
"""Functions and classes related to file and directory handling
"""
import os
import shutil
import subprocess
import sys
import atexit
import codecs
import numpy as np
import json
import json_tricks
try:
import cPickle as pickle
except ImportError:
import pickle
from psychopy import logging
from psychopy.tools.fileerrortools import handleFileCollision
from pathlib import Path
def _synonymiseExtensions(assets):
"""
Synonymise filetypes which refer to the same media types.
Parameters
==========
assets : dict
Dict of {name: path} pairs
Returns
==========
dict
Same dict which was passed in, but any names ending in a recognised extension
will have variants with the same stem but different (and synonymous) extensions,
pointing to the same path. For example:
{"default.png": "default.png"}
becomes
{"default.png": "default.png", "default.jpg": "default.png", "default.jpeg": "default.png"}
"""
# Alias filetypes
newAssets = {}
for key, val in assets.items():
# Skip if no ext
if "." not in key:
continue
# Get stem and ext
stem, ext = key.split(".")
# Synonymise image types
imgTypes = ("png", "jpg", "jpeg")
if ext in imgTypes:
for thisExt in imgTypes:
newAssets[stem + "." + thisExt] = val
# Synonymise movie types
movTypes = ("mp4", "mov", "mkv", "avi", "wmv")
if ext in movTypes:
for thisExt in movTypes:
newAssets[stem + "." + thisExt] = val
# Synonymise audio types
sndTypes = ("mp3", "wav")
if ext in sndTypes:
for thisExt in sndTypes:
newAssets[stem + "." + thisExt] = val
return newAssets
# Names accepted by stimulus classes & the filename of the default stimulus to use
defaultStimRoot = Path(__file__).parent.parent / "assets"
defaultStim = {
# Image stimuli
"default.png": "default.png",
# Movie stimuli
"default.mp4": "default.mp4",
# Sound stimuli
"default.mp3": "default.mp3",
# Credit card image
"creditCard.png": "creditCard.png",
"CreditCard.png": "creditCard.png",
"creditcard.png": "creditCard.png",
# USB image
"USB.png": "USB.png",
"usb.png": "USB.png",
# USB-C image
"USB-C.png": "USB-C.png",
"USB_C.png": "USB-C.png",
"USBC.png": "USB-C.png",
"usb-c.png": "USB-C.png",
"usb_c.png": "USB-C.png",
"usbc.png": "USB-C.png",
}
defaultStim = _synonymiseExtensions(defaultStim)
[docs]def toFile(filename, data):
"""Save data (of any sort) as a pickle file.
simple wrapper of the cPickle module in core python
"""
f = open(filename, 'wb')
pickle.dump(data, f)
f.close()
[docs]def fromFile(filename, encoding='utf-8-sig'):
"""Load data from a psydat, pickle or JSON file.
Parameters
----------
encoding : str
The encoding to use when reading a JSON file. This parameter will be
ignored for any other file type.
"""
filename = pathToString(filename)
if filename.endswith('.psydat'):
with open(filename, 'rb') as f:
try:
contents = pickle.load(f)
except UnicodeDecodeError:
f.seek(0) # reset to start of file to try again
contents = pickle.load(f, encoding='latin1') # python 2 data files
# if loading an experiment file make sure we don't save further
# copies using __del__
if hasattr(contents, 'abort'):
contents.abort()
return contents
elif filename.endswith('pickle'):
with open(filename, 'rb') as f:
contents = pickle.load(f)
return contents
elif filename.endswith('.json'):
with codecs.open(filename, 'r', encoding=encoding) as f:
contents = json_tricks.load(f)
# Restore RNG if we load a TrialHandler2 object.
# We also need to remove the 'temporary' ._rng_state attribute that
# was saved with it.
from psychopy.data import TrialHandler2
if isinstance(contents, TrialHandler2):
contents._rng = np.random.default_rng()
contents._rng.bit_generator.state = contents._rng_state
del contents._rng_state
return contents
# QuestPlus.
if sys.version_info.major == 3 and sys.version_info.minor >= 6:
from psychopy.data.staircase import QuestPlusHandler
from questplus import QuestPlus
if isinstance(contents, QuestPlusHandler):
# Restore the questplus.QuestPlus object.
contents._qp = QuestPlus.from_json(contents._qp_json)
del contents._qp_json
return contents
# If we haven't returned anything by now, the loaded object is neither
# a TrialHandler2 nor a QuestPlus instance. Return it unchanged.
return contents
else:
msg = "Don't know how to handle this file type, aborting."
raise ValueError(msg)
[docs]def mergeFolder(src, dst, pattern=None):
"""Merge a folder into another.
Existing files in `dst` folder with the same name will be
overwritten. Non-existent files/folders will be created.
"""
# dstdir must exist first
srcnames = os.listdir(src)
for name in srcnames:
srcfname = os.path.join(src, name)
dstfname = os.path.join(dst, name)
if os.path.isdir(srcfname):
if not os.path.isdir(dstfname):
os.makedirs(dstfname)
mergeFolder(srcfname, dstfname)
else:
try:
# copy without metadata:
shutil.copyfile(srcfname, dstfname)
except IOError as why:
print(why)
[docs]def openOutputFile(fileName=None, append=False, fileCollisionMethod='rename',
encoding='utf-8-sig'):
"""Open an output file (or standard output) for writing.
:Parameters:
fileName : None, 'stdout', or str
The desired output file name. If `None` or `stdout`, return
`sys.stdout`. Any other string will be considered a filename.
append : bool, optional
If ``True``, append data to an existing file; otherwise, overwrite
it with new data.
Defaults to ``True``, i.e. appending.
fileCollisionMethod : string, optional
How to handle filename collisions. Valid values are `'rename'`,
`'overwrite'`, and `'fail'`.
This parameter is ignored if ``append`` is set to ``True``.
Defaults to `rename`.
encoding : string, optional
The encoding to use when writing the file. This parameter will be
ignored if `append` is `False` and `fileName` ends with `.psydat`
or `.npy` (i.e. if a binary file is to be written).
Defaults to ``'utf-8'``.
:Returns:
f : file
A writable file handle.
"""
fileName = pathToString(fileName)
if (fileName is None) or (fileName == 'stdout'):
return sys.stdout
if append:
mode = 'a'
else:
if fileName.endswith(('.psydat', '.npy')):
mode = 'wb'
else:
mode = 'w'
# Rename the output file if a file of that name already exists
# and it should not be appended.
if os.path.exists(fileName) and not append:
fileName = handleFileCollision(
fileName,
fileCollisionMethod=fileCollisionMethod)
# Do not use encoding when writing a binary file.
if 'b' in mode:
encoding = None
if os.path.exists(fileName) and mode in ['w', 'wb']:
logging.warning('Data file %s will be overwritten!' % fileName)
# The file will always be opened in binary writing mode,
# see https://docs.python.org/2/library/codecs.html#codecs.open
f = codecs.open(fileName, mode=mode, encoding=encoding)
return f
[docs]def genDelimiter(fileName):
"""Return a delimiter based on a filename.
:Parameters:
fileName : string
The output file name.
:Returns:
delim : string
A delimiter picked based on the supplied filename. This will be
``,`` if the filename extension is ``.csv``, and a tabulator
character otherwise.
"""
fileName = pathToString(fileName)
if fileName.endswith(('.csv', '.CSV')):
delim = ','
else:
delim = '\t'
return delim
def genFilenameFromDelimiter(filename, delim):
# If no known filename extension was specified, derive a one from the
# delimiter.
filename = pathToString(filename)
if not filename.endswith(('.dlm', '.DLM', '.tsv', '.TSV', '.txt',
'.TXT', '.csv', '.CSV', '.psydat', '.npy',
'.json')):
if delim == ',':
filename += '.csv'
elif delim == '\t':
filename += '.tsv'
else:
filename += '.txt'
return filename
def constructLegacyFilename(filename):
# make path object from filename
filename = Path(filename)
# construct legacy variant name
legacyName = filename.parent / (filename.stem + "_legacy" + filename.suffix)
return legacyName
class DictStorage(dict):
"""Helper class based on dictionary with storage to json
"""
def __init__(self, filename, *args, **kwargs):
dict.__init__(self, *args, **kwargs)
self.filename = filename
self.load()
self._deleted = False
atexit.register(self.__del__)
def load(self, filename=None):
"""Load all tokens from a given filename
(defaults to ~/.PsychoPy3/pavlovia.json)
"""
if filename is None:
filename = self.filename
if os.path.isfile(filename):
with open(filename, 'r') as f:
try:
self.update(json.load(f))
except ValueError:
logging.error("Tried to load %s but it wasn't valid "
"JSON format"
%filename)
def save(self, filename=None):
"""Save all tokens from a given filename
(defaults to the filename given to the class but can be overridden)
"""
if filename is None:
filename = self.filename
# make sure the folder exists
folder = os.path.split(filename)[0]
if not os.path.isdir(folder):
os.makedirs(folder)
# save the file as json
with open(filename, 'wb') as f:
json_str = json.dumps(self, indent=2, sort_keys=True)
f.write(bytes(json_str, 'UTF-8'))
def __del__(self):
if not self._deleted:
self.save()
self._deleted = True
class KnownProjects(DictStorage):
def save(self, filename=None):
"""Purge unnecessary projects (without a local root) and save"""
toPurge = []
for projname in self:
proj = self[projname]
if not proj['localRoot']:
toPurge.append(projname)
for projname in toPurge:
del self[projname]
DictStorage.save(self, filename)
def pathToString(filepath):
"""
Coerces pathlib Path objects to a string (only python version 3.6+)
any other objects passed to this function will be returned as is.
This WILL NOT work with on Python 3.4, 3.5 since the __fspath__ under
method did not exist in those versions, however psychopy does not support
these versions of python anyways.
:Parameters:
filepath : str or pathlib.Path
file system path that needs to be coerced into a string to
use by Psychopy's internals
:Returns:
filepath : str or same as input object
file system path coerced into a string type
"""
if hasattr(filepath, "__fspath__"):
return filepath.__fspath__()
return filepath
def openInExplorer(path):
"""
Open a given director path in current operating system's file explorer.
"""
# Choose a command according to OS
if sys.platform in ['win32']:
comm = "explorer"
elif sys.platform in ['darwin']:
comm = "open"
elif sys.platform in ['linux', 'linux2']:
comm = "dolphin"
# Use command to open folder
ret = subprocess.call(" ".join([comm, path]), shell=True)
return ret
