Commit cef935fe authored by Mark Hymers's avatar Mark Hymers Committed by Joe Lyons
Browse files

Add start of MEG transfer code


Signed-off-by: Mark Hymers's avatarMark Hymers <mark.hymers@hankel.co.uk>
parent bfb22d41
......@@ -47,6 +47,16 @@ class YIASPhysioConversionError(YIASError):
__all__.append('YIASPhysioConversionError')
class YIASAnonError(YIASError):
# Holds some error information
def __init__(self, err): # pragma: nocover
self.err = err
def __str__(self): # pragma: nocover
return 'YIASAnonError: (%s)' % (self.err)
__all__.append('YIASAnonError')
class YIASSeriesPluginError(YIASError):
# Holds some error information
def __init__(self, err): # pragma: nocover
......@@ -66,3 +76,32 @@ class YIASStudyPluginError(YIASError):
return 'YIASStudyPluginError: (%s)' % (self.err)
__all__.append('YIASStudyPluginError')
class YIASMissingFileError(YIASError):
def __init__(self, filename):
self.filename = filename
def __str__(self): # pragma: nocover
return 'YIASMissingFileError: (%s)' % (self.filename)
__all__.append('YIASMissingFileError')
class YIASNotMEGHDF5Error(YIASError):
def __init__(self, filename, details):
self.filename = filename
self.details = details
def __str__(self): # pragma: nocover
return 'YIASNotMEGHDF5Error: (%s, %s)' % (self.filename, self.details)
__all__.append('YIASNotMEGHDF5Error')
class YIASMEGHDF5MetaMissing(YIASError):
def __init__(self, filename, details):
self.filename = filename
self.details = details
def __str__(self): # pragma: nocover
return 'YIASMEGHDF5MetaMissing: (%s, %s)' % (self.filename, self.details)
__all__.append('YIASMEGHDF5MetaMissing')
#!/usr/bin/python3
# New module for YNiC imaging archive service code
from copy import deepcopy
from os import chmod, unlink
from os.path import join, basename, isfile, dirname
from shutil import copy2
from subprocess import call
from tempfile import mkdtemp, NamedTemporaryFile
import dateutil.parser
import h5py
from .errors import (YIASNoSeriesError, YIASMissingFileError,
YIASNotMEGHDF5Error, YIASMEGHDF5MetaMissing)
from .utils import sanitise_string
__all__ = []
def hdf5_is_megscan(filename):
"""
Quick check whether an HDF5 file is a MEGSCAN format one and that
we recognise the version.
Raises YIASNotMEGHDF5Error on problem
"""
try:
with h5py.File(filename, 'r') as f:
if f.attrs.get('format_family', None) != 'MEGSCAN':
raise Exception('format_family is missing or not MEGSCAN')
if f.attrs.get('format_version', None) != 'rev9':
raise Exception('format_version is missing or not rev9')
except Exception as e:
raise YIASNotMEGHDF5Error(filename, str(e))
return True
def anonymise_hdf5(filename, anonid):
"""
Anonymise the relevant fields in an HDF5 file (filename).
The file must be closed when this is performed as an h5repack will
have to be performed on the file
"""
with NamedTemporaryFile(dir=dirname(filename), suffix='hdf5') as tfile:
tmpname1 = tfile.name
copy2(filename, tfile.name)
f = h5py.File(tfile.name, 'a')
f['subject'].attrs['id'] = anonid
f['subject'].attrs['name'] = anonid
f['subject'].attrs['dob'] = ''
f['subject'].attrs['sex'] = ''
f.close()
# We now need to repack the file to ensure all of the old data is gone
with NamedTemporaryFile(dir=dirname(filename), suffix='hdf5') as tfile:
tmpname2 = tfile.name
if call(['h5repack', tmpname1, tmpname2]) != 0: # pragma: nocover
raise YIASAnonError('Failed to run h5repack')
# Copy the file to the final location
copy2(tmpname2, filename)
__all__.append('anonymise_hdf5')
class MEGIncomingHandler(object):
"""
This class is responsible for parsing an MEG upload containing
one raw and/or one processed file
"""
def __init__(self, raw_file, proc_file):
# At the moment we need both files
if raw_file is None and proc_file is None:
raise YIASNoSeriesError()
# Check raw file
if raw_file is not None:
if not isfile(raw_file):
raise YIASMissingFileError(raw_file)
# This will raise on an error
hdf5_is_megscan(raw_file)
self.raw_file = raw_file
# Check proc file
if proc_file is not None:
if not isfile(proc_file):
raise YIASMissingFileError(proc_file)
# This will raise on an error
hdf5_is_megscan(proc_file)
self.proc_file = proc_file
# Prepare some useful values
self._read_metainfo()
def _read_metainfo(self):
"""
Read in:
+ Participant ID
+ Protocol name
+ Scan date/time
"""
# TODO: Might want to check the values against both files if present
if self.proc_file is not None:
filename = self.proc_file
else:
filename = self.raw_file
try:
with h5py.File(filename, 'r') as f:
part_id = f['subject'].attrs['id']
# Problem: protocol name is only available in the processed file
# We can write a GUI to fix this for raw-only cases before upload
protocol_name = f['config']['protocol'].attrs['protocol_name']
# Look for the start time of the first acquisition
protocol_time = f['acquisitions']['0'].attrs['start_time']
# Convert the time to a usable datetime
protocol_time = dateutil.parser.parse(protocol_time)
except Exception as e:
raise YIASMEGHDF5MetaMissing(filename, str(e))
self.part_id = part_id
self.protocol_name = protocol_name
self.protocol_time = protocol_time
__all__.append('MEGIncomingHandler')
#!/usr/bin/python3
# This script runs tests on the hdf5.py file routines
from os import stat, symlink, listdir, unlink, makedirs
from os.path import join, isdir, isfile
from shutil import copyfile
import h5py
from tempfile import TemporaryDirectory
import pytest
from .test_support import (TEST_BASEDIR, BaseCopyTest)
########################################################################
# HDF5 tests
########################################################################
class TestAnonymisation(BaseCopyTest):
inputs_dirs = ['meg/realscan1']
def test_anonymisation(self, tmpdir):
from yias.meghdf5 import anonymise_hdf5
from yias.errors import YIASNotMEGHDF5Error
proc_file = join(tmpdir, 'incoming', 'processed', '2019-11-01T121555.388380+0000_R1000_MEG.hdf5')
# Take a copy of the original file to work on
new_file = join(tmpdir, 'tmp.hdf5')
copyfile(proc_file, new_file)
anonymise_hdf5(new_file, 'R9999')
with h5py.File(new_file, 'r') as f:
assert f['subject'].attrs['id'] == 'R9999'
assert f['subject'].attrs['name'] == 'R9999'
assert f['subject'].attrs['sex'] == ''
assert f['subject'].attrs['dob'] == ''
# We also need to check that R1000 doesn't appear anywhere in the raw file
# i.e. that we've repacked the data out of it
f = open(new_file, 'rb')
data = f.read()
with pytest.raises(ValueError):
data.index(b'R1000')
class TestCaseIncomingProcessorBadData(BaseCopyTest):
inputs_dirs = ['meg/baddata1']
def test_create_incoming_handler_nodata(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
from yias.errors import YIASNotMEGHDF5Error
raw_file = join(tmpdir, 'incoming', 'raw', 'notrealdata.hdf5')
proc_file = join(tmpdir, 'incoming', 'processed', 'notrealdata.hdf5')
with pytest.raises(YIASNotMEGHDF5Error):
ih = MEGIncomingHandler(raw_file, proc_file)
def test_create_incoming_handler_missing_files(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
from yias.errors import YIASMissingFileError
raw_file = join(tmpdir, 'incoming', 'raw', 'NOTAFILE.hdf5')
proc_file = join(tmpdir, 'incoming', 'raw', 'NOTAFILE.hdf5')
with pytest.raises(YIASMissingFileError):
ih = MEGIncomingHandler(raw_file, None)
with pytest.raises(YIASMissingFileError):
ih = MEGIncomingHandler(None, proc_file)
class TestCaseIncomingProcessorBadData2(BaseCopyTest):
inputs_dirs = ['meg/baddata2']
def test_create_incoming_handler_nodata(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
from yias.errors import YIASNotMEGHDF5Error
raw_file = join(tmpdir, 'incoming', 'raw', 'notrealdata.hdf5')
proc_file = join(tmpdir, 'incoming', 'processed', 'notrealdata.hdf5')
with pytest.raises(YIASNotMEGHDF5Error):
ih = MEGIncomingHandler(raw_file, proc_file)
class TestCaseIncomingProcessor(BaseCopyTest):
inputs_dirs = ['meg/realscan1']
def test_create_incoming_handler_nodata(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
from yias.errors import YIASNoSeriesError
with pytest.raises(YIASNoSeriesError):
ih = MEGIncomingHandler(None, None)
def test_create_incoming_handler_realdata_raw_only(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
from yias.errors import YIASMEGHDF5MetaMissing
raw_file = join(tmpdir, 'incoming', 'raw', '2019-11-01T121555.388380+0000_R1000_MEG.hdf5')
# At the moment this fails as the raw file doesn't have the protocol in
# We'll provide a tool to add this information before upload
# so we are correct in rejecting this
with pytest.raises(YIASMEGHDF5MetaMissing):
ih = MEGIncomingHandler(raw_file, None)
def test_create_incoming_handler_realdata_proc_only(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
proc_file = join(tmpdir, 'incoming', 'processed', '2019-11-01T121555.388380+0000_R1000_MEG.hdf5')
ih = MEGIncomingHandler(None, proc_file)
assert(ih.part_id == 'R1000')
assert(ih.protocol_name == 'P1401a')
assert(ih.protocol_time.year == 2019)
assert(ih.protocol_time.month == 11)
assert(ih.protocol_time.day == 1)
def test_create_incoming_handler_realdata_both(self, tmpdir):
from yias.meghdf5 import MEGIncomingHandler
raw_file = join(tmpdir, 'incoming', 'raw', '2019-11-01T121555.388380+0000_R1000_MEG.hdf5')
proc_file = join(tmpdir, 'incoming', 'processed', '2019-11-01T121555.388380+0000_R1000_MEG.hdf5')
ih = MEGIncomingHandler(raw_file, proc_file)
assert(ih.part_id == 'R1000')
assert(ih.protocol_name == 'P1401a')
assert(ih.protocol_time.year == 2019)
assert(ih.protocol_time.month == 11)
assert(ih.protocol_time.day == 1)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment