RAMControl/run.py at master · melkalliny/RAMControl · 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
105
106
107
108
109
110
111
112
113
"""Script to start RAM experiments."""

from __future__ import print_function

import os
import pickle
from configparser import ConfigParser
from argparse import ArgumentParser
import re
import subprocess

from prompt_toolkit import prompt
from prompt_toolkit.history import InMemoryHistory
from prompt_toolkit.contrib.completers import WordCompleter

import ramcontrol
from ramcontrol.util import absjoin


def get_subject(subject=""):
    """Prompt for the subject ID.

    :param str subject: Default option.

    """
    history = InMemoryHistory()
    validate = lambda s: len(re.findall(r"R\d{4}[A-Z]", s)) == 1
    while True:
        # response = prompt(u"Subject [{}]: ".format(subject), history=history)
        response = prompt(u"Subject: ")
        if len(response) == 0 and validate(subject):
            return subject.encode()
        elif validate(response):
            return response.encode()
        else:
            subject = response
            print("Invalid subject ID")


def get_experiment(available, experiment=""):
    """Prompt for the experiment to run.

    :param list available: Available experiments.
    :param str experiment: Default choice.

    """
    completer = WordCompleter(available)
    while True:
        # response = prompt(u"Experiment (press tab to see available) [{}]: ".format(experiment),
        #                   completer=completer, complete_while_typing=True)
        response = prompt(u"Experiment (press tab to see available): ",
                          completer=completer, complete_while_typing=True)
        if len(response) == 0 and experiment in available:
            return experiment.encode()
        elif response in available:
            return response.encode()
        else:
            experiment = ""
            print("Invalid experiment")


def main():
    config = ConfigParser()
    config.read("ramcontrol.ini")

    parser = ArgumentParser()
    parser.add_argument("-s", "--subject", help="Subject ID", default=None)
    parser.add_argument("-x", "--experiment", default=None, help="Experiment to run")
    parser.add_argument("-d", "--debug", action="store_true", default=False,
                        help="Enable debug mode")
    parser.add_argument("--no-fs", action="store_true", default=False,
                        help="Disable fullscreen mode")
    args = parser.parse_args()

    experiments = config["general"]["experiments"].split()

    if args.debug:
        print("!"*80)
        print("{:^80}".format("DEBUG MODE ENABLED!"))
        print("{:^80}".format("Restart NOW if this is a real experiment!!!"))
        print("!"*80)

    if args.subject is None:
        args.subject = get_subject()
    if args.experiment is None:
        args.experiment = get_experiment(experiments)

    env = {
        "subject": args.subject,
        "experiment": args.experiment,
        "experiment_family": config.get(args.experiment, "family"),

        "voiceserver": config[args.experiment].getboolean("voiceserver", fallback=False),

        "video_path": os.path.abspath(os.path.expanduser(config["videos"]["path"])),
        "data_path": absjoin("./data"),
        "ramcontrol_path": os.path.dirname(ramcontrol.__file__),

        "fullscreen": not (args.debug or args.no_fs),
        "debug": args.debug,
        "debug_options": config["debug"].items()
    }

    penv = os.environ.copy()
    penv["RAM_CONFIG"] = pickle.dumps(env)

    p = subprocess.Popen(["python", "-m", "ramcontrol.experiment"],
                         cwd=absjoin("."), env=penv)
    p.wait()


if __name__ == "__main__":
    main()