Skip to content

Executors

plink1Executor

Bases: object

A wrapper class for interfacing with the plink1.9 command line tool.

Source code in magenpy/utils/executors.py 
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
class plink1Executor(object):
    """
    A wrapper class for interfacing with the `plink1.9` command line tool.
    """

    def __init__(self, threads=None, verbose=True):
        """
        Initialize the plink1.9 executor
        :param threads: The number of threads to use for computations. If None, the number of
        threads will be set to 1.
        :type threads: int or str
        :param verbose: Whether to print the output of the command
        :type verbose: bool
        """

        self.threads = threads or 1

        self.plink1_path = get_option('plink1.9_path')

        if not is_cmd_tool(self.plink1_path):
            raise Exception(f"Did not find the executable for plink1.9 at: {self.plink1_path}")

        self.verbose = verbose

    def execute(self, cmd):
        """
        Execute a plink command
        :param cmd: The flags to pass to plink. For example, ['--bfile', 'file', '--out', 'output']
        :type cmd: list of strings
        :raises: CalledProcessError if the command fails
        """

        cmd = [self.plink1_path] + cmd + [f'--threads {self.threads}']

        from subprocess import CalledProcessError

        try:
            run_shell_script(" ".join(cmd))
        except CalledProcessError as e:
            if self.verbose:
                print("Invocation of plink returned the following error message:")
                print(e.stderr.decode())
            raise e

__init__(threads=None, verbose=True)

Initialize the plink1.9 executor

Parameters:

Name Type Description Default
threads int | str

The number of threads to use for computations. If None, the number of threads will be set to 1.

 None
verbose bool

Whether to print the output of the command

 True
Source code in magenpy/utils/executors.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, threads=None, verbose=True):
    """
    Initialize the plink1.9 executor
    :param threads: The number of threads to use for computations. If None, the number of
    threads will be set to 1.
    :type threads: int or str
    :param verbose: Whether to print the output of the command
    :type verbose: bool
    """

    self.threads = threads or 1

    self.plink1_path = get_option('plink1.9_path')

    if not is_cmd_tool(self.plink1_path):
        raise Exception(f"Did not find the executable for plink1.9 at: {self.plink1_path}")

    self.verbose = verbose

execute(cmd)

Execute a plink command

Parameters:

Name Type Description Default
cmd list of strings

The flags to pass to plink. For example, ['--bfile', 'file', '--out', 'output']

 required
Source code in magenpy/utils/executors.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def execute(self, cmd):
    """
    Execute a plink command
    :param cmd: The flags to pass to plink. For example, ['--bfile', 'file', '--out', 'output']
    :type cmd: list of strings
    :raises: CalledProcessError if the command fails
    """

    cmd = [self.plink1_path] + cmd + [f'--threads {self.threads}']

    from subprocess import CalledProcessError

    try:
        run_shell_script(" ".join(cmd))
    except CalledProcessError as e:
        if self.verbose:
            print("Invocation of plink returned the following error message:")
            print(e.stderr.decode())
        raise e

plink2Executor

Bases: object

A wrapper class for interfacing with the plink2 command line tool.

Source code in magenpy/utils/executors.py 
 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
class plink2Executor(object):
    """
    A wrapper class for interfacing with the `plink2` command line tool.
    """

    def __init__(self, threads=None, verbose=True):
        """
        Initialize the plink2 executor
        :param threads: The number of threads to use for computations. If None, the number of
        threads will be set to 1.
        :type threads: int or str
        :param verbose: Whether to print the output of the command
        :type verbose: bool
        """

        self.threads = threads or 1

        self.plink2_path = get_option('plink2_path')

        if not is_cmd_tool(self.plink2_path):
            raise Exception(f"Did not find the executable for plink2 at: {self.plink2_path}")

        self.verbose = verbose

    def execute(self, cmd):
        """
        Execute a `plink2` command
        :param cmd: The flags to pass to plink2. For example, ['--bfile', 'file', '--out', 'output']
        :type cmd: list of strings
        :raises CalledProcessError: If the command returns a non-zero exit code
        """

        cmd = [self.plink2_path] + cmd + [f'--threads {self.threads}']

        from subprocess import CalledProcessError

        try:
            run_shell_script(" ".join(cmd))
        except CalledProcessError as e:

            if self.verbose:
                print("Invocation of plink2 returned the following error message:")
                print(e.stderr.decode())

            raise e

__init__(threads=None, verbose=True)

Initialize the plink2 executor

Parameters:

Name Type Description Default
threads int | str

The number of threads to use for computations. If None, the number of threads will be set to 1.

 None
verbose bool

Whether to print the output of the command

 True
Source code in magenpy/utils/executors.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def __init__(self, threads=None, verbose=True):
    """
    Initialize the plink2 executor
    :param threads: The number of threads to use for computations. If None, the number of
    threads will be set to 1.
    :type threads: int or str
    :param verbose: Whether to print the output of the command
    :type verbose: bool
    """

    self.threads = threads or 1

    self.plink2_path = get_option('plink2_path')

    if not is_cmd_tool(self.plink2_path):
        raise Exception(f"Did not find the executable for plink2 at: {self.plink2_path}")

    self.verbose = verbose

execute(cmd)

Execute a plink2 command

Parameters:

Name Type Description Default
cmd list of strings

The flags to pass to plink2. For example, ['--bfile', 'file', '--out', 'output']

 required

Raises:

Type Description
CalledProcessError

If the command returns a non-zero exit code
Source code in magenpy/utils/executors.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def execute(self, cmd):
    """
    Execute a `plink2` command
    :param cmd: The flags to pass to plink2. For example, ['--bfile', 'file', '--out', 'output']
    :type cmd: list of strings
    :raises CalledProcessError: If the command returns a non-zero exit code
    """

    cmd = [self.plink2_path] + cmd + [f'--threads {self.threads}']

    from subprocess import CalledProcessError

    try:
        run_shell_script(" ".join(cmd))
    except CalledProcessError as e:

        if self.verbose:
            print("Invocation of plink2 returned the following error message:")
            print(e.stderr.decode())

        raise e