An easy-to-use platform for EEG experimentation in the classroom
Merge pull request #25 from neurosity/master
@neurosity/pipes upgrade and misc
authored by
Dano Morrison
and committed by
GitHub
eb520e74
778f2da4
+17
-53
+1
-1
app/package.json
+1
-1
app/package.json
+5
-22
app/utils/eeg/pipes.js
+5
-22
app/utils/eeg/pipes.js
···
1
-
import { Observable } from 'rxjs';
1
+
import { pipe } from 'rxjs';
2
2
import { map } from 'rxjs/operators';
3
3
import {
4
4
SIGNAL_QUALITY,
5
5
SIGNAL_QUALITY_THRESHOLDS
6
6
} from '../../constants/constants';
7
7
8
-
export const parseMuseSignalQuality = () => source =>
9
-
createPipe(
10
-
source,
8
+
export const parseMuseSignalQuality = () =>
9
+
pipe(
11
10
map(epoch => ({
12
11
...epoch,
13
12
signalQuality: Object.assign(
···
30
29
}))
31
30
);
32
31
33
-
export const parseEmotivSignalQuality = () => source =>
34
-
createPipe(
35
-
source,
32
+
export const parseEmotivSignalQuality = () =>
33
+
pipe(
36
34
map(epoch => ({
37
35
...epoch,
38
36
signalQuality: Object.assign(
···
54
52
)
55
53
}))
56
54
);
57
-
58
-
const createPipe = (source, ...pipes) =>
59
-
new Observable(observer =>
60
-
source.pipe(...pipes).subscribe({
61
-
next(event) {
62
-
observer.next(event);
63
-
},
64
-
error(err) {
65
-
observer.error(err);
66
-
},
67
-
complete() {
68
-
observer.complete();
69
-
}
70
-
})
71
-
);
+11
-30
app/utils/jupyter/pipes.js
+11
-30
app/utils/jupyter/pipes.js
···
1
-
import { Observable } from "rxjs";
2
-
import { map, pluck, filter, take, mergeMap } from "rxjs/operators";
3
-
import { executeRequest } from "@nteract/messaging";
4
-
import { RECEIVE_EXECUTE_REPLY } from "../../epics/jupyterEpics";
1
+
import { pipe } from 'rxjs';
2
+
import { map, pluck, filter, take, mergeMap } from 'rxjs/operators';
3
+
import { executeRequest } from '@nteract/messaging';
4
+
import { RECEIVE_EXECUTE_REPLY } from '../../epics/jupyterEpics';
5
5
6
6
// Refactor this so command can be calculated either up stream or inside pipe
7
-
export const execute = (command, state$) => source =>
8
-
createPipe(
9
-
source,
10
-
map(() =>
11
-
state$.value.jupyter.mainChannel.next(executeRequest(command))
12
-
)
7
+
export const execute = (command, state$) =>
8
+
pipe(
9
+
map(() => state$.value.jupyter.mainChannel.next(executeRequest(command)))
13
10
);
14
11
15
-
export const awaitOkMessage = action$ => source =>
16
-
createPipe(
17
-
source,
12
+
export const awaitOkMessage = action$ =>
13
+
pipe(
18
14
mergeMap(() =>
19
15
action$.ofType(RECEIVE_EXECUTE_REPLY).pipe(
20
-
pluck("payload"),
21
-
filter(msg => msg.channel === "shell" && msg.content.status === "ok"),
16
+
pluck('payload'),
17
+
filter(msg => msg.channel === 'shell' && msg.content.status === 'ok'),
22
18
take(1)
23
19
)
24
20
)
25
21
);
26
-
27
-
const createPipe = (source, ...pipes) =>
28
-
new Observable(observer =>
29
-
source.pipe(...pipes).subscribe({
30
-
next(event) {
31
-
observer.next(event);
32
-
},
33
-
error(err) {
34
-
observer.error(err);
35
-
},
36
-
complete() {
37
-
observer.complete();
38
-
}
39
-
})
40
-
);