snirf.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Jun  3 13:28:35 2020

@author: theodorehuppert
"""

import h5py as h5py
import numpy as np
import re
import colorama
from colorama import Fore, Style
import sys

class data:
    dataTimeSeries = None
    time = None
    measurementList=None

class measurementList:
    sourceIndex=None
    detectorIndex=None
    wavelengthIndex=None
    wavelengthActual=None
    wavelengthEmissionsActual=None
    dataType=None
    dataTypeLabel=None
    dataTypeIndex=None
    sourcePower=None
    detectorGain=None
    moduleIndex=None

class probe:
    wavelengths=None
    wavelengthsEmission=None
    sourcePos2D=None
    sourcePos3D=None
    detectorPos2D=None
    detectorPos3D=None
    frequencies=None
    timeDelays=None
    timeDelayWidths=None
    momentOrder=None
    correlationTimeDelays=None
    correlationTimeDelayWidths=None
    sourceLabels=None
    detectorLabels=None
    landmarkPos2D=None
    landmarkPos3D=None
    landmarkLabels=None

class aux:
    name=None
    dataTimeSeries=None
    time=None
    timeOffSet=None

class stim:
    name=None
    data=None

class snirf:
    filename=None
    formatVersion = "1.0"
    data=None
    stim=None
    probe=None
    aux=None
    metaDataTags=None

def read_from_file(filename):
     fileID = h5py.File(filename, 'r')
     formatVersion=hdfgetdata(fileID,"/formatVersion")
     keys=fileID['/']
     s=[]
     for x in keys:
         if 'nirs' in x:
             a=snirf
             a.formatVersion=formatVersion
             a.probe=read_probe(fileID['/'][x]['probe']);
             a.data=[];
             a.aux=[];
             a.stim=[];
             keys2=fileID['/'][x]
             for x2 in keys2:
                 if 'data' in x2:
                     d = read_data(fileID['/'][x][x2])
                     a.data.append(d)
                 if 'aux' in x2:
                     ax = read_aux(fileID['/'][x][x2])
                     a.aux.append(ax)
                 if 'stim' in x2:
                     ax = read_stim(fileID['/'][x][x2])
                     a.stim.append(ax)

             a.stim=np.array(a.stim)
             a.data=np.array(a.data)
             a.aux=np.array(a.aux)
             s.append(a)
     if len(s)==1:
         s=s[0]

     return s

def read_data(gID):
    d = data

    d.dataTimeSeries =hdfgetdata(gID,'dataTimeSeries')
    d.time=hdfgetdata(gID,'time')
    d.measurementList=[];
    for fld in gID:
        if 'measurementList' in fld:
            d.measurementList=read_measurementList(gID[fld])

    d.measurementList=np.array(measurementList)

    return d

def read_probe(gID):
    p=probe

    p.wavelengths=hdfgetdata(gID,'wavelengths')
    p.wavelengthsEmission=hdfgetdata(gID,'wavelengthsEmission')
    p.sourcePos2D=hdfgetdata(gID,'sourcePos2D')
    p.sourcePos3D=hdfgetdata(gID,'sourcePos3D')
    p.detectorPos2D=hdfgetdata(gID,'detectorPos2D')
    p.detectorPos3D=hdfgetdata(gID,'detectorPos3D')
    p.frequencies=hdfgetdata(gID,'frequencies')
    p.timeDelays=hdfgetdata(gID,'timeDelays')
    p.timeDelayWidths=hdfgetdata(gID,'timeDelayWidths')
    p.momentOrder=hdfgetdata(gID,'momentOrder')
    p.correlationTimeDelays=hdfgetdata(gID,'correlationTimeDelays')
    p.correlationTimeDelayWidths=hdfgetdata(gID,'correlationTimeDelayWidths')
    p.sourceLabels=hdfgetdata(gID,'sourceLabels')
    p.detectorLabels=hdfgetdata(gID,'detectorLabels')
    p.landmarkPos2D=hdfgetdata(gID,'landmarkPos2D')
    p.landmarkPos3D=hdfgetdata(gID,'landmarkPos3D')
    p.landmarkLabels=hdfgetdata(gID,'landmarkLabels')


    return p

def read_aux(gID):
    a=aux
    a.name=hdfgetdata(gID,'name')
    a.dataTimeSeries=hdfgetdata(gID,'dataTimeSeries')
    a.time=hdfgetdata(gID,'time')
    a.timeOffSet=hdfgetdata(gID,'timeOffSet')

    return a

def read_stim(gID):
    s=stim
    s.name=hdfgetdata(gID,'name')
    s.data=hdfgetdata(gID,'data')
    return s

def read_measurementList(gID):
    m=measurementList



    m.sourceIndex=hdfgetdata(gID,'sourceIndex')
    m.detectorIndex=hdfgetdata(gID,'detectorIndex')
    m.wavelengthIndex=hdfgetdata(gID,'wavelengthIndex')
    m.wavelengthActual=hdfgetdata(gID,'wavelengthActual')
    m.wavelengthEmissionsActual=hdfgetdata(gID,'wavelengthEmissionsActual')
    m.dataType=hdfgetdata(gID,'dataType')
    m.dataTypeLabel=hdfgetdata(gID,'dataTypeLabel')
    m.dataTypeIndex=hdfgetdata(gID,'dataTypeIndex')
    m.sourcePower=hdfgetdata(gID,'sourcePower')
    m.detectorGain=hdfgetdata(gID,'detectorGain')
    m.moduleIndex=hdfgetdata(gID,'moduleIndex')

    return m

def hdfgetdata(gID,field):
        val = gID.get(field)
        if val is None:
            return val

        if h5py.check_string_dtype(val.dtype):
            # string
            if val.len()==1:
                val=val[0].tobytes().decode('ascii')
                return val
            else:
                val2=[];
                for x in val:
                    val2.append(x.tobytes().decode('ascii'))
                val2=np.array(val2)
                return val2
        val=np.array(val)

        if(val.ndim==1 and len(val)==1):
            val=val[0]

        return val

def getrequiredfieldsLst():
     Required=[];
     Required.append("/formatVersion")
     Required.append("/nirs\d*/metaDataTags/SubjectID")
     Required.append("/nirs\d*/metaDataTags/MeasurementDate")
     Required.append("/nirs\d*/metaDataTags/MeasurementTime")
     Required.append("/nirs\d*/metaDataTags/LengthUnit")
     Required.append("/nirs\d*/metaDataTags/TimeUnit")
     Required.append("/nirs\d*/metaDataTags/FrequencyUnit")
     Required.append("/nirs\d*/data\d*/dataTimeSeries")
     Required.append("/nirs\d*/data\d*/time")
     Required.append("/nirs\d*/data\d*/measurementList\d*/sourceIndex")
     Required.append("/nirs\d*/data\d*/measurementList\d*/detectorIndex")
     Required.append("/nirs\d*/data\d*/measurementList\d*/wavelengthIndex")
     Required.append("/nirs\d*/data\d*/measurementList\d*/dataType")
     Required.append("/nirs\d*/data\d*/measurementList\d*/dataTypeIndex")
     Required.append("/nirs\d*/probe/wavelengths")
     Required.append("/nirs\d*/probe/sourcePos2D")
     Required.append("/nirs\d*/probe/detectorPos2D")
     Required.append("/nirs\d*/stim\d*/name")
     Required.append("/nirs\d*/stim\d*/data")
     Required.append("/nirs\d*/aux\d*/name")
     Required.append("/nirs\d*/aux\d*/dataTimeSeries")
     Required.append("/nirs\d*/aux\d*/time")
     return Required

def getoptionalfieldsLst():
     Optional=[]
     Optional.append("/nirs\d*/data\w*/measurementList\d*/wavelengthActual")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/wavelengthEmissionActual")
     Optional.append("/nirs\d*/data\d*/measurementList\d*/dataTypeLabel")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/sourcePower")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/detectorGain")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/moduleIndex")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/sourceModuleIndex")
     Optional.append("/nirs\d*/data\w*/measurementList\d*/detectorModuleIndex")
     Optional.append("/nirs\d*/probe/wavelengthsEmission")
     Optional.append("/nirs\d*/probe/sourcePos3D")
     Optional.append("/nirs\d*/probe/detectorPos3D")
     Optional.append("/nirs\d*/probe/frequencies")
     Optional.append("/nirs\d*/probe/timeDelays")
     Optional.append("/nirs\d*/probe/timeDelayWidths")
     Optional.append("/nirs\d*/probe/momentOrders")
     Optional.append("/nirs\d*/probe/correlationTimeDelays")
     Optional.append("/nirs\d*/probe/correlationTimeDelayWidths")
     Optional.append("/nirs\d*/probe/sourceLabels")
     Optional.append("/nirs\d*/probe/detectorLabels")
     Optional.append("/nirs\d*/probe/landmarkPos2D")
     Optional.append("/nirs\d*/probe/landmarkPos3D")
     Optional.append("/nirs\d*/probe/landmarkLabels")
     Optional.append("/nirs\d*/probe/useLocalIndex")
     Optional.append("/nirs\d*/aux\d*/timeOffset")
     Optional.append("/nirs\d*/stim\d*/dataLabels")
     Optional.append("/nirs\d*/metaDataTags/ManufacturerName")
     Optional.append("/nirs\d*/metaDataTags/ManufacturerName")
     Optional.append("/nirs\d*/metaDataTags/Model")
     Optional.append("/nirs\d*/metaDataTags/SubjectName")
     Optional.append("/nirs\d*/metaDataTags/DateOfBirth")
     Optional.append("/nirs\d*/metaDataTags/AcquisitionStartTime")
     Optional.append("/nirs\d*/metaDataTags/StudyID")
     Optional.append("/nirs\d*/metaDataTags/StudyDescription")
     Optional.append("/nirs\d*/metaDataTags/AccessionNumber")
     Optional.append("/nirs\d*/metaDataTags/InstanceNumber")
     Optional.append("/nirs\d*/metaDataTags/CalibrationFileName")
     Optional.append("/nirs\d*/metaDataTags/UnixTime")
     return Optional

def isrequired(fld):
    flag = False
    required=getrequiredfieldsLst()
    for x in required:
        if re.match(x,fld):
            flag = True
    return flag

def isoptional(fld):
    flag = False
    required=getoptionalfieldsLst()
    for x in required:
        if re.match(x,fld):
            flag = True
    return flag

def validate(filename,fileOut=None):
     fileID = h5py.File(filename, 'r')
     formatVersion=hdfgetdata(fileID,"/formatVersion")

     def getallnames(gID,lst):
         if isinstance(gID, h5py.Dataset):
             lst.append(gID.name)
         else:
            for x in gID:
                getallnames(gID[x],lst)

     def checkdim(field,fID,foundInvalid,lstInvalid):
         val = fID.get(field);

         if "Pos2D" in field or "Pos3D" in field:
             dim = 2
         elif "dataTimeSeries" in field:
             dim = 2
         else:
             dim = 1
         if dim != len(val.dims):
             return False

     lst=[]

     getallnames(fileID,lst)

     if fileOut == None:
         print('-' * 40)
         print('SNIRF Validator')
         print('Version 1.0')
         print('written by T. Huppert')
         print()
         print('File = {0}'.format(filename))
         print('Version = {0}'.format(formatVersion))
         print('-' * 40)

         foundInvalid=0;

         lstInvalid=[]

         for x in lst:
            print(Fore.WHITE + x)
            val = fileID.get(x)
            if h5py.check_string_dtype(val.dtype):
                # string
                if val.len()==1:
                    val=val[0].tobytes().decode('ascii')
                    print('\tHDF5-STRING: {0}'.format(val))
                else:
                    val2=[];
                    for y in val:
                        val2.append(y.tobytes().decode('ascii'))
                    val2=np.array(val2)
                    print('\tHDF5-STRING 1D-Vector: <{0}x1>'.format(len(val2)))
            else:
                val=np.array(val)
                if(val.ndim==1 and len(val)==1):
                    val=val[0]
                    print('\tHDF5-FLOAT: {0}'.format(val))
                elif val.ndim==1:

                     print('\tHDF5-FLOAT 1D-Vector: <{0}x1>'.format(len(val)))
                else:
                     print('\tHDF5-FLOAT 2D-Array: <{0}x{1}>'.format(len(val),int(val.size/len(val))))

            dimcheck = checkdim(x, fileID, foundInvalid, lstInvalid)
            if dimcheck == False:
                val = len(fileID.get(x).dims)
                if val == 1:
                    print(Fore.RED +'\tINVALID dimensions(Expected Number of Dimensions: 2)')
                else:
                    print(Fore.RED +'\tINVALID dimensions(Expected Number of Dimensions: 1)')
                foundInvalid=foundInvalid+1;
                lstInvalid.append(x)

            if "/aux" in x or "/stim" in x:
                if isrequired(x) == True:
                    print(Fore.BLUE + '\t\tRequired field when optional parent object is included')
                elif isoptional(x):
                    print(Fore.GREEN + '\t\tOptional field when optional parent object is included')
                else:
                    print(Fore.RED + '\t\tINVALID field')
                    foundInvalid = foundInvalid + 1
                    lstInvalid.append(x)
            else:
                if isrequired(x) == True:
                    print(Fore.BLUE + '\t\tRequired field')
                elif isoptional(x):
                    print(Fore.GREEN + '\t\tOptional field')
                else:
                    print(Fore.RED + '\t\tINVALID field')
                    foundInvalid = foundInvalid + 1
                    lstInvalid.append(x)

         print('-' * 40)
         if(len(lstInvalid)!=0):
              print(Fore.RED+ "File is INVALID")
              print(Fore.RED +'\tINVALID ENTRIES FOUND')
              for x in lstInvalid:
                  print(Fore.RED + x)
         else:
              print(Fore.WHITE+ "File is VALID")
         print(Style.RESET_ALL)
     else: # write to file
         text_file = open(fileOut, "w")
         text_file.write('\n' + '\n' + '-' * 40)
         text_file.write('\n' + '\n' + 'SNIRF Validator')
         text_file.write('\n' + '\n' + 'Version 1.0')
         text_file.write('\n' + 'written by T. Huppert')
         text_file.write('\n')
         text_file.write('\n' + 'File = {0}'.format(filename))
         text_file.write('\n' + 'Version = {0}'.format(formatVersion))
         text_file.write('\n' + '-' * 40)

         foundInvalid=0;

         lstInvalid=[]

         for x in lst:
            text_file.write('\n' + x)
            val = fileID.get(x)
            if h5py.check_string_dtype(val.dtype):
                # string
                if val.len()==1:
                    val=val[0].tobytes().decode('ascii')
                    text_file.write('\n' + '\tHDF5-STRING: {0}'.format(val))
                else:
                    val2=[];
                    for y in val:
                        val2.append(y.tobytes().decode('ascii'))
                    val2=np.array(val2)
                    text_file.write('\n' + '\tHDF5-STRING 1D-Vector: <{0}x1>'.format(len(val2)))
            else:
                val=np.array(val)
                if(val.ndim==1 and len(val)==1):
                    val=val[0]
                    text_file.write('\n' + '\tHDF5-FLOAT: {0}'.format(val))
                elif val.ndim==1:

                     text_file.write('\n' + '\tHDF5-FLOAT 1D-Vector: <{0}x1>'.format(len(val)))
                else:
                     text_file.write('\n' + '\tHDF5-FLOAT 2D-Array: <{0}x{1}>'.format(len(val),int(val.size/len(val))))

            dimcheck = checkdim(x, fileID, foundInvalid, lstInvalid)
            if dimcheck == False:
                val = len(fileID.get(x).dims)
                if val == 1:
                    text_file.write(Fore.RED +'\tINVALID dimensions(Expected Number of Dimensions: 2)')
                else:
                    text_file.write(Fore.RED +'\tINVALID dimensions(Expected Number of Dimensions: 1)')
                foundInvalid=foundInvalid+1
                lstInvalid.append(x)

            if isrequired(x)==True:
                text_file.write('\n' + '\t\tRequired field')
            elif isoptional(x):
                text_file.write('\n' + '\t\tOptional field')
            else:
                text_file.write('\n' + '\t\tINVALID field')
                foundInvalid=foundInvalid+1
                lstInvalid.append(x)

         text_file.write('\n' + '-' * 40)
         if(len(lstInvalid)!=0):
              text_file.write('\n' + "File is INVALID")
              text_file.write('\n' + '\tINVALID ENTRIES FOUND')
              for x in lstInvalid:
                  text_file.write('\n' +  x)
         else:
              text_file.write('\n' + "File is VALID")
         text_file.close()
     return (foundInvalid==0)

def main():

    filename=sys.argv[1]
    print(filename)
    if(len(sys.argv)>2):
        fileOut=sys.argv[2]
    else:
        fileOut=None
    validate(filename,fileOut)

if __name__ == "__main__":
    main()