Skip to content

Use os.path instead of string operations in EEG #1342

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cmadjar merged 2 commits into from
Nov 28, 2025
Merged

Use os.path instead of string operations in EEG #1342

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
30de8ed
use os.path.join in eeg
MaximeBICMTL Nov 21, 2025
901ab28
also change path in physiological.py
MaximeBICMTL Nov 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 48 additions & 49 deletions python/lib/eeg.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,29 +310,29 @@ def register_data(self, derivatives=False, detect=True):
)

# archive all files in a tar ball for downloading all files at once
files_to_archive = (os.path.join(self.data_dir, eeg_file_path),)
files_to_archive: list[str] = [os.path.join(self.data_dir, eeg_file_path)]

if eegjson_file_path:
files_to_archive = (*files_to_archive, os.path.join(self.data_dir, eegjson_file_path))
files_to_archive.append(os.path.join(self.data_dir, eegjson_file_path))
if fdt_file_path:
files_to_archive = (*files_to_archive, os.path.join(self.data_dir, fdt_file_path))
files_to_archive.append(os.path.join(self.data_dir, fdt_file_path))
if electrode_file_path:
files_to_archive = (*files_to_archive, os.path.join(self.data_dir, electrode_file_path))
files_to_archive.append(os.path.join(self.data_dir, electrode_file_path))
if event_file_paths:
# archive all event files in a tar ball for event download
event_files_to_archive = ()
event_files_to_archive: list[str] = []

for event_file_path in event_file_paths:
files_to_archive = (*files_to_archive, os.path.join(self.data_dir, event_file_path))
event_files_to_archive = (*event_files_to_archive, os.path.join(self.data_dir, event_file_path))
files_to_archive.append(os.path.join(self.data_dir, event_file_path))
event_files_to_archive.append(os.path.join(self.data_dir, event_file_path))

event_archive_rel_name = os.path.splitext(event_file_paths[0])[0] + ".tgz"
self.create_and_insert_event_archive(
event_files_to_archive, event_archive_rel_name, eeg_file_id
)

if channel_file_path:
files_to_archive = (*files_to_archive, os.path.join(self.data_dir, channel_file_path))
files_to_archive.append(os.path.join(self.data_dir, channel_file_path))

archive_rel_name = os.path.splitext(eeg_file_path)[0] + ".tgz"
self.create_and_insert_archive(
Expand Down Expand Up @@ -416,7 +416,7 @@ def fetch_and_insert_eeg_files(self, derivatives=False, detect=True):
with open(eegjson_file.path) as data_file:
eeg_file_data = json.load(data_file)

eegjson_file_path = eegjson_file.path.replace(self.data_dir, '')
eegjson_file_path = os.path.relpath(eegjson_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the JSON file to the LORIS BIDS import directory
eegjson_file_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -456,7 +456,7 @@ def fetch_and_insert_eeg_files(self, derivatives=False, detect=True):
# eeg_file_data dictionary
fdt_file_path = None
if file_type == 'set' and fdt_file:
fdt_file_path = fdt_file.path.replace(self.data_dir, '')
fdt_file_path = os.path.relpath(fdt_file, self.data_dir)
if self.loris_bids_root_dir:
# copy the fdt file to the LORIS BIDS import directory
fdt_file_path = self.copy_file_to_loris_bids_dir(
Expand All @@ -483,7 +483,7 @@ def fetch_and_insert_eeg_files(self, derivatives=False, detect=True):
# grep the modality ID from physiological_modality table
modality_id = physiological_modality.grep_id_from_modality_value(self.bids_modality)

eeg_path = eeg_file.path.replace(self.data_dir, '')
eeg_path = os.path.relpath(eeg_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the eeg_file to the LORIS BIDS import directory
eeg_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -577,7 +577,7 @@ def fetch_and_insert_electrode_file(
)
if not result:
electrode_data = utilities.read_tsv_file(electrode_file.path)
electrode_path = electrode_file.path.replace(self.data_dir, '')
electrode_path = os.path.relpath(electrode_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the electrode file to the LORIS BIDS import directory
electrode_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -617,7 +617,7 @@ def fetch_and_insert_electrode_file(
electrode_ids
)
else:
electrode_metadata_path = coordsystem_metadata_file.path.replace(self.data_dir, '')
electrode_metadata_path = os.path.relpath(coordsystem_metadata_file, self.data_dir)
if self.loris_bids_root_dir:
# copy the electrode metadata file to the LORIS BIDS import directory
electrode_metadata_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -686,7 +686,7 @@ def fetch_and_insert_channel_file(
channel_path = result[0]['FilePath'] if result else None
channel_data = utilities.read_tsv_file(channel_file.path)
if not result:
channel_path = channel_file.path.replace(self.data_dir, '')
channel_path = os.path.relpath(channel_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the channel file to the LORIS BIDS import directory
channel_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -771,7 +771,7 @@ def fetch_and_insert_event_files(
'with physiological file ID ' + str(physiological_file_id)
print(message)
else:
event_metadata_path = event_metadata_file.path.replace(self.data_dir, '')
event_metadata_path = os.path.relpath(event_metadata_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the event file to the LORIS BIDS import directory
event_metadata_path = self.copy_file_to_loris_bids_dir(
Expand All @@ -796,7 +796,7 @@ def fetch_and_insert_event_files(

# get events.tsv file and insert
event_data = utilities.read_tsv_file(event_data_file.path)
event_path = event_data_file.path.replace(self.data_dir, '')
event_path = event_path = os.path.relpath(event_data_file.path, self.data_dir)
if self.loris_bids_root_dir:
# copy the event file to the LORIS BIDS import directory
event_path = self.copy_file_to_loris_bids_dir(
Expand Down Expand Up @@ -840,56 +840,48 @@ def copy_file_to_loris_bids_dir(self, file, derivatives=False, inheritance=False
# Handle derivatives differently
# Data path structure is unpredictable, so keep the same relative path
if derivatives:
copy_file = str.replace(
file,
self.bids_layout.root,
""
)
copy_file = self.loris_bids_root_dir + copy_file

# create derivative directories
lib.utilities.create_dir(
os.path.dirname(copy_file),
self.verbose
)
copy_file = os.path.relpath(file, self.bids_layout.root)
copy_file = os.path.join(self.loris_bids_root_dir, copy_file)
else :
# determine the path of the copied file
copy_file = ""
if not inheritance:
copy_file = self.loris_bids_eeg_rel_dir
if self.bids_ses_id:
copy_file += os.path.basename(file)
copy_file = os.path.join(copy_file, os.path.basename(file))
else:
# make sure the ses- is included in the new filename if using
# default visit label from the LORIS config
copy_file += str.replace(
os.path.basename(file),
"sub-" + self.bids_sub_id,
"sub-" + self.bids_sub_id + "_ses-" + self.default_vl
copy_file = os.path.join(
copy_file,
os.path.basename(file).replace(
"sub-" + self.data_type.subject.label,
"sub-" + self.data_type.subject.label + "_ses-" + self.default_vl
)
)
copy_file = self.loris_bids_root_dir + copy_file

# create the directory if it does not exist
lib.utilities.create_dir(
os.path.dirname(copy_file),
self.verbose
)

# copy the file
utilities.copy_file(file, copy_file, self.verbose)

# determine the relative path and return it
relative_path = copy_file.replace(self.data_dir, "")
relative_path = os.path.relpath(copy_file, self.data_dir)

return relative_path

def create_and_insert_archive(self, files_to_archive, archive_rel_name,
eeg_file_id):
def create_and_insert_archive(self, files_to_archive: list[str], archive_rel_name: str, eeg_file_id: int):
"""
Create an archive with all electrophysiology files associated to a
specific recording (including electrodes.tsv, channels.tsv etc...)

:param files_to_archive: tuple with the list of files to include in
the archive
:type files_to_archive: tuple
:param files_to_archive: list of files to include in the archive
:param archive_rel_name: path to the archive relative to data_dir
:type archive_rel_name: str
:param eeg_file_id : PhysiologicalFileID
:type eeg_file_id : int
"""

# load the Physiological object that will be used to insert the
Expand Down Expand Up @@ -923,6 +915,12 @@ def create_and_insert_archive(self, files_to_archive, archive_rel_name,
else:
return

# create the archive directory if it does not exist
lib.utilities.create_dir(
os.path.dirname(archive_full_path),
self.verbose
)

# create the archive file
utilities.create_archive(files_to_archive, archive_full_path)

Expand All @@ -935,17 +933,12 @@ def create_and_insert_archive(self, files_to_archive, archive_rel_name,
}
physiological.insert_archive_file(archive_info)

def create_and_insert_event_archive(self, files_to_archive, archive_rel_name, eeg_file_id):
def create_and_insert_event_archive(self, files_to_archive: list[str], archive_rel_name: str, eeg_file_id: int):
"""
Create an archive with all event files associated to a specific recording

:param files_to_archive: tuple with the list of files to include in
the archive
:type files_to_archive: tuple
:param files_to_archive: list of files to include in the archive
:param archive_rel_name: path to the archive relative to data_dir
:type archive_rel_name: str
:param eeg_file_id : PhysiologicalFileID
:type eeg_file_id : int
"""

# check if archive is on the filesystem
Expand Down Expand Up @@ -978,6 +971,12 @@ def create_and_insert_event_archive(self, files_to_archive, archive_rel_name, ee
else:
return

# create the archive directory if it does not exist
lib.utilities.create_dir(
os.path.dirname(archive_full_path),
self.verbose
)

# create the archive file
utilities.create_archive(files_to_archive, archive_full_path)

Expand Down
18 changes: 11 additions & 7 deletions python/lib/physiological.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import sys
from dataclasses import dataclass
from functools import reduce
from pathlib import Path

import lib.exitcode
from lib.database_lib.bids_event_mapping import BidsEventMapping
Expand Down Expand Up @@ -1225,17 +1226,20 @@ def create_chunks_for_visualization(self, physio_file_id, data_dir):
if not chunk_root_dir:
# the bids_rel_dir is the first two directories in file_path (
# bids_imports/BIDS_dataset_name_BIDSVersion)
bids_rel_dir = file_path.split('/')[0] + '/' + file_path.split('/')[1]
chunk_root_dir = data_dir + bids_rel_dir + '_chunks' + '/'
file_path_components = Path(file_path).parts
bids_rel_dir = os.path.join(file_path_components[0], file_path_components[1])
chunk_root_dir = os.path.join(data_dir, f'{bids_rel_dir}_chunks')

full_file_path = os.path.join(data_dir, file_path)

# determine which script to run based on the file type
file_type = self.grep_file_type_from_file_id(physio_file_id)
if file_type == 'set':
script = os.environ['LORIS_MRI'] + '/python/loris_eeg_chunker/eeglab_to_chunks.py'
command = 'python ' + script + ' ' + data_dir + file_path + ' --destination ' + chunk_root_dir
command = 'python ' + script + ' ' + full_file_path + ' --destination ' + chunk_root_dir
elif file_type == 'edf':
script = os.environ['LORIS_MRI'] + '/python/loris_eeg_chunker/edf_to_chunks.py'
command = 'python ' + script + ' ' + data_dir + file_path + ' --destination ' + chunk_root_dir
command = 'python ' + script + ' ' + full_file_path + ' --destination ' + chunk_root_dir

# chunk the electrophysiology dataset if a command was determined above
try:
Expand All @@ -1251,11 +1255,11 @@ def create_chunks_for_visualization(self, physio_file_id, data_dir):
print('ERROR: ' + script + ' not found')
sys.exit(lib.exitcode.CHUNK_CREATION_FAILURE)

chunk_path = chunk_root_dir + os.path.splitext(os.path.basename(file_path))[0] + '.chunks'
chunk_path = os.path.join(chunk_root_dir, os.path.splitext(os.path.basename(file_path))[0] + '.chunks')
if os.path.isdir(chunk_path):
self.insert_physio_parameter_file(
physiological_file_id = physio_file_id,
parameter_name = 'electrophysiology_chunked_dataset_path',
value = chunk_path.replace(chunk_root_dir_config, '') if chunk_root_dir_config
else chunk_path.replace(data_dir, '')
value = os.path.relpath(chunk_path, chunk_root_dir_config) if chunk_root_dir_config
else os.path.relpath(chunk_path, data_dir)
)
Loading