brainflow/emulator/brainflow_emulator/freeeeg32_windows.py at 15146f9b6efc9a52d5c17f1e4a5d79655ddef3da · brainflow-dev/brainflow · GitHub

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
96
97
98
99
100
101
102
103
104
import logging
import os
import subprocess
import sys
import time

from pathlib import Path
from brainflow_emulator.emulate_common import TestFailureError, log_multilines
from brainflow_emulator.freeeeg32_emulator import Listener
from serial import Serial


def write(port, data):
    return port.write(data)


def read(port, num_bytes):
    return port.read(num_bytes)


def get_installer():
    return str(
        Path(__file__).resolve().parent
        / "com0com"
        / "setup_com0com_W7_x64_signed.exe"
    )


def install_com0com():
    this_directory = os.path.abspath(os.path.dirname(__file__))
    directory = os.path.join(this_directory, 'com0com')
    if not os.path.exists(directory):
        os.makedirs(directory)
    cmds = [get_installer(), '/NCRC', '/S', '/D=%s' % directory]
    logging.info('running %s' % ' '.join(cmds))
    p = subprocess.Popen(cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = p.communicate()
    if p.returncode != 0:
        logging.error('stdout is %s' % out)
        logging.error('stderr is %s' % err)
        raise Exception('com0com installation failure')
    logging.info('Sleeping a few second, it doesnt work in appveyour without it')
    time.sleep(10)
    return directory


def get_ports_windows():
    directory = install_com0com()
    # remove ports from previous run if any
    p = subprocess.Popen([os.path.join(directory, 'setupc.exe'), 'remove', '0'],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=directory)
    stdout, stderr = p.communicate()
    logging.info('remove stdout is %s' % stdout)
    logging.info('remove stderr is %s' % stderr)

    m_name = 'COM14'
    s_name = 'COM15'

    p = subprocess.Popen(
        [os.path.join(directory, 'setupc.exe'), 'install', 'PortName=%s' % m_name, 'PortName=%s' % s_name],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=directory)
    stdout, stderr = p.communicate()
    logging.info('install stdout is %s' % stdout)
    logging.info('install stderr is %s' % stderr)

    if p.returncode != 0:
        raise Exception('com0com failure')
    logging.info('Sleeping a few second, it doesnt work in appveyour without it')
    time.sleep(10)
    return m_name, s_name


def test_serial(cmd_list, m_name, s_name):
    master = Serial('\\\\.\\%s' % m_name, timeout=0)
    listen_thread = Listener(master, write, read)
    listen_thread.daemon = True
    listen_thread.start()

    cmd_to_run = cmd_list + [s_name]
    logging.info('Running %s' % ' '.join([str(x) for x in cmd_to_run]))
    process = subprocess.Popen(cmd_to_run, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = process.communicate()

    log_multilines(logging.info, stdout)
    log_multilines(logging.info, stderr)

    master.close()
    if process.returncode != 0:
        raise TestFailureError('Test failed with exit code %s' % str(process.returncode), process.returncode)

    return stdout, stderr


def main(cmd_list):
    if not cmd_list:
        raise Exception('No command to execute')

    m_name, s_name = get_ports_windows()
    test_serial(cmd_list, m_name, s_name)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    main(sys.argv[1:])