(self, code, stream_handler=None)
return TextOutput(output)
Execute the code in the subprocess.
        Use the stream_handler to process lines as they are emitted
        by the subprocess.
Execute the code in the subprocess.
        Use the stream_handler to process lines as they are emitted
        by the subprocess.
Execute the code in the subprocess . Use the stream_handler to process lines as they are emitted by the subprocess .
def do_execute_direct_single_command_expred(self, code, stream_handler=None):
        """Execute the code in the subprocess.
        Use the stream_handler to process lines as they are emitted
        by the subprocess.
        """
        self.payload = []

        if not code.strip():
            self.kernel_resp = {
                "status": "ok",
                "execution_count": self.execution_count,
                "payload": [],
                "user_expressions": {},
            }
            return

        interrupted = False
        output = ""
        try:
            output = self.wrapper.run_command(
                code.rstrip(), timeout=-1, stream_handler=stream_handler
            )
            # TODO:  instead of proccess_response, it would be better
            # to capture the prints and warnings at the moment they are sended.
            output = self.process_response(output)
            if stream_handler is not None:
                output = ""

        except MMASyntaxError as e:
            self.kernel_resp = {
                "status": "error",
                "execution_count": self.execution_count,
                "ename": e.name,
                "evalue": e.val,
                "traceback": e.traceback,
            }
            return TextOutput("null:")
        except KeyboardInterrupt as e:
            interrupted = True
            output = self.wrapper.child.before
            if "REPL not responding to interrupt" in str(e):
                output += "\n%s" % e
        except EOF:
            output = self.wrapper.child.before + "Restarting"
            self._start()

        if interrupted:
            self.kernel_resp = {
                "status": "abort",
                "execution_count": self.execution_count,
            }

        exitcode, trace = self.check_exitcode()

        if exitcode:
            self.kernel_resp = {
                "status": "error",
                "execution_count": self.execution_count,
                "ename": "",
                "evalue": str(exitcode),
                "traceback": trace,
            }
        else:
            self.kernel_resp = {
                "status": "ok",
                "execution_count": self.execution_count,
                "payload": [],
                "user_expressions": {},
            }

        return TextOutput(output)
def do_execute_direct_single_command_expred ( self , code , stream_handler = None ) : self . payload = [ ] if not code . strip ( ) : self . kernel_resp = { "status" : "ok" , "execution_count" : self . execution_count , "payload" : [ ] , "user_expressions" : { } , } return interrupted = False output = "" try : output = self . wrapper . run_command ( code . rstrip ( ) , timeout = - 1 , stream_handler = stream_handler ) # TODO:  instead of proccess_response, it would be better # to capture the prints and warnings at the moment they are sended. output = self . process_response ( output ) if stream_handler is not None : output = "" except MMASyntaxError as e : self . kernel_resp = { "status" : "error" , "execution_count" : self . execution_count , "ename" : e . name , "evalue" : e . val , "traceback" : e . traceback , } return TextOutput ( "null:" ) except KeyboardInterrupt as e : interrupted = True output = self . wrapper . child . before if "REPL not responding to interrupt" in str ( e ) : output += "\n%s" % e except EOF : output = self . wrapper . child . before + "Restarting" self . _start ( ) if interrupted : self . kernel_resp = { "status" : "abort" , "execution_count" : self . execution_count , } exitcode , trace = self . check_exitcode ( ) if exitcode : self . kernel_resp = { "status" : "error" , "execution_count" : self . execution_count , "ename" : "" , "evalue" : str ( exitcode ) , "traceback" : trace , } else : self . kernel_resp = { "status" : "ok" , "execution_count" : self . execution_count , "payload" : [ ] , "user_expressions" : { } , } return TextOutput ( output )