FilesExpand file tree

main

/

trca_benchmark.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

95 lines (77 loc) · 2.92 KB

main

/

trca_benchmark.py

Copy path

Top

File metadata and controls

• Code
• Blame

95 lines (77 loc) · 2.92 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

# for running locally

import os

cwd = os.getcwd()

import sys

# path = os.path.join(cwd, "..\\..\\")

path = cwd

sys.path.append(path)

# imports

import numpy as np

import logging

logging.getLogger('lightning').setLevel(0)

import warnings

warnings.filterwarnings('ignore')

from splearn.data import MultipleSubjects, Benchmark

from splearn.utils import Logger, Config

from splearn.filter.butterworth import butter_bandpass_filter

from splearn.filter.notch import notch_filter

from splearn.filter.channels import pick_channels

from splearn.cross_decomposition.trca import TRCA

from splearn.cross_validate.leave_one_out import block_evaluation

config = {

"experiment_name": "trcaEnsemble_benchmark",

"data": {

"load_subject_ids": np.arange(1,36),

"root": "../data/hsssvep",

"selected_channels": ["PZ", "PO5", "PO3", "POz", "PO4", "PO6", "O1", "Oz", "O2"],

"duration": 1,

},

"trca": {

"ensemble": True

},

"seed": 1234

}

main_logger = Logger(filename_postfix=config["experiment_name"])

main_logger.write_to_log("Config")

main_logger.write_to_log(config)

config = Config(config)

# define custom preprocessing steps

def func_preprocessing(data):

data_x = data.data

data_x = pick_channels(data_x, channel_names=data.channel_names, selected_channels=config.data.selected_channels)

data_x = notch_filter(data_x, sampling_rate=data.sampling_rate, notch_freq=50.0)

data_x = butter_bandpass_filter(data_x, lowcut=7, highcut=90, sampling_rate=data.sampling_rate, order=6)

start_t = 35

end_t = start_t + (config.data.duration * data.sampling_rate)

data_x = data_x[:,:,:,start_t:end_t]

data.set_data(data_x)

# load data

data = MultipleSubjects(

dataset=Benchmark,

root=os.path.join(path,config.data.root),

subject_ids=config.data.load_subject_ids,

func_preprocessing=func_preprocessing,

verbose=True,

)

num_channel = data.data.shape[2]

num_classes = data.stimulus_frequencies.shape[0]

signal_length = data.data.shape[3]

def leave_one_block_evaluation(classifier, X, Y, block_seq_labels=None):

test_results_acc = []

blocks, targets, channels, samples = X.shape

main_logger.write_to_log("Begin", break_line=True)

for block_i in range(blocks):

test_acc = block_evaluation(classifier, X, Y, block_i)

test_results_acc.append(test_acc)

this_result = {

"test_subject_id": block_i+1,

"acc": test_acc,

}

main_logger.write_to_log(this_result)

mean_acc = np.array(test_results_acc).mean().round(3)*100

print(f'Mean test accuracy: {mean_acc}%')

main_logger.write_to_log("Mean acc: "+str(mean_acc), break_line=True)

trca_classifier = TRCA(sampling_rate=data.sampling_rate, ensemble=config.trca.ensemble)

print("data:", data.data.shape)

print("targets:", data.targets.shape)

leave_one_block_evaluation(classifier=trca_classifier, X=data.data, Y=data.targets)