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
114
115
116
117
118
119
|
# Copyright 2016 Free Software Foundation, Inc.
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-2.0-or-later
#
from __future__ import absolute_import
import os
import shlex
import subprocess
import threading
import time
from distutils.spawn import find_executable
from gi.repository import GLib
from ..core import Messages
class ExecFlowGraphThread(threading.Thread):
"""Execute the flow graph as a new process and wait on it to finish."""
def __init__(self, flow_graph_page, xterm_executable, callback):
"""
ExecFlowGraphThread constructor.
"""
threading.Thread.__init__(self)
self.page = flow_graph_page # store page and don't use main window calls in run
self.flow_graph = self.page.flow_graph
self.xterm_executable = xterm_executable
self.update_callback = callback
try:
if self.flow_graph.get_option('output_language') == 'python':
self.process = self.page.process = self._popen()
elif self.flow_graph.get_option('output_language') == 'cpp':
self.process = self.page.process = self._cpp_popen()
self.update_callback()
self.start()
except Exception as e:
Messages.send_verbose_exec(str(e))
Messages.send_end_exec()
def _popen(self):
"""
Execute this python flow graph.
"""
generator = self.page.get_generator()
run_command = self.flow_graph.get_run_command(generator.file_path)
run_command_args = shlex.split(run_command)
# When in no gui mode on linux, use a graphical terminal (looks nice)
xterm_executable = find_executable(self.xterm_executable)
if generator.generate_options == 'no_gui' and xterm_executable:
run_command_args = [xterm_executable, '-e', run_command]
# this does not reproduce a shell executable command string, if a graphical
# terminal is used. Passing run_command though shlex_quote would do it but
# it looks really ugly and confusing in the console panel.
Messages.send_start_exec(' '.join(run_command_args))
return subprocess.Popen(
args=run_command_args,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
shell=False, universal_newlines=True
)
def _cpp_popen(self):
"""
Execute this C++ flow graph after generating and compiling it.
"""
generator = self.page.get_generator()
run_command = generator.file_path + '/build/' + self.flow_graph.get_option('id')
dirname = generator.file_path
builddir = os.path.join(dirname, 'build')
if os.path.isfile(run_command):
os.remove(run_command)
xterm_executable = find_executable(self.xterm_executable)
run_command_args = ['cmake .. &&', 'make && ', xterm_executable, '-e', run_command]
Messages.send_start_exec(' '.join(run_command_args))
return subprocess.Popen(
args=' '.join(run_command_args),
cwd=builddir,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
shell=True, universal_newlines=True
)
def run(self):
"""
Wait on the executing process by reading from its stdout.
Use GObject.idle_add when calling functions that modify gtk objects.
"""
# handle completion
r = "\n"
while r:
GLib.idle_add(Messages.send_verbose_exec, r)
r = self.process.stdout.read(1)
# Properly close pipe before thread is terminated
self.process.stdout.close()
while self.process.poll() is None:
# Wait for the process to fully terminate
time.sleep(0.05)
GLib.idle_add(self.done)
def done(self):
"""Perform end of execution tasks."""
Messages.send_end_exec(self.process.returncode)
self.page.process = None
self.update_callback()
|
