<commit_before>import os
import sys
import cmd
import argparse
import runner
import topics
import pytelemetry.pytelemetry as tm
import pytelemetry.transports.serialtransport as transports
from ui import Plot
import numpy as np

class Console(cmd.Cmd):

    def __init__(self):
        cmd.Cmd.__init__(self)

        self.transport = transports.SerialTransport()
        self.telemetry = tm.pytelemetry(self.transport)
        self.topics = topics.Topics(self.telemetry)
        self.runner = runner.Runner(self.transport,self.telemetry)

        self.completekey = None
        self.prompt = ">> "
        self.intro  = "pytelemetry terminal started."  ## defaults to None

    ## Command definitions ##
    def do_ls(self, args):
        """
            Prints a list of received topics
        """
        t = self.topics.ls()
        for i in t:
            print(i)

    def do_print(self, args):
        """
            Prints X last received samples on a given topic
        """

        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        parser.add_argument('--amount','-a', dest='amount', type=int, default=10)
        args = parser.parse_args(args.split())

        s = self.topics.samples(args.topic,args.amount)
        if s is not None:
            for i in s:
                print(i)

    def do_plot(self, args):
        """
            Plots a topic in a PyQtGraph plot
        """
        if not self.topics.exists(args):
            print("Topic [",args,"] unknown.")

        print("Plotting:", args)

        plt = Plot.Plot2D()
        data = self.topics.samples(args,amount=0)
        data[3] = 12
        data[5] = -4
        print(data)
        t = np.arange(0,len(data))
        print(t)
        plt.trace(args,t,data)
        plt.start()

    def do_count(self, args):
        """Counts amount of received samples on a given topic"""

        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        args = parser.parse_args(args.split())

        s = self.topics.samples(args.topic,args.amount)
        for i in s:
            print(i)

    def do_exit(self, args):
        """Exits from the terminal"""
        return -1

    def do_connect(self,args):
        """Connect to a serial port"""
        # Parse args
        parser = argparse.ArgumentParser()
        parser.add_argument('--port','-p', dest='port', required=True)
        parser.add_argument('--baudrate','-b', dest='baudrate', type=int, default=115200)
        args = parser.parse_args(args.split())

        # Build the option dictionnary
        options = dict()
        options['port'] = args.port
        options['baudrate'] = args.baudrate

        try:
            self.runner.connect(options)
        except:
            print("Failed to connect to :",options['port']," at ",options['baudrate']," (bauds)")
            print("Connection error : ",sys.exc_info())
        else:
            print("Connected to :",options['port']," at ",options['baudrate']," (bauds)")

    def do_disconnect(self,args):
        self.runner.disconnect()
        print("Disconnected.")

    def do_pub(self,args):
        """Publish data on a topic"""
        # Parse args
        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        parser.add_argument('--data','-d', dest='data', required=True)
        parser.add_argument('--type', dest='type', required=True)

        try:
            args = parser.parse_args(args.split())
        except Exception as e:
            print("Error :",e)
            return

        valid_types = ['string','uint8','uint16','uint32','int8','int16','int32','float32']

        if not args.type in valid_types:
            print("Invalid type. Not in ",valid_types)
            return

        if args.type == 'float32':
            args.data = float(args.data)
        elif args.type != 'string':
            args.data = int(args.data)

        print("Published on |",args.topic,"|",args.data,"[",args.type,"]")

        self.telemetry.publish(args.topic,args.data,args.type)

    ## Command definitions to support Cmd object functionality ##
    def do_help(self, args):
        """Get help on commands
           'help' or '?' with no arguments prints a list of commands for which help is available
           'help <command>' or '? <command>' gives help on <command>
        """
        ## The only reason to define this method is for the help text in the doc string
        cmd.Cmd.do_help(self, args)

    ## Override methods in Cmd object ##
    def preloop(self):
        """Initialization before prompting user for commands.
           Despite the claims in the Cmd documentaion, Cmd.preloop() is not a stub.
        """
        cmd.Cmd.preloop(self)   ## sets up command completion
        self._hist    = []      ## No history yet
        self._locals  = {}      ## Initialize execution namespace for user
        self._globals = {}

    def postloop(self):
        """Take care of any unfinished business.
           Despite the claims in the Cmd documentaion, Cmd.postloop() is not a stub.
        """
        cmd.Cmd.postloop(self)   ## Clean up command completion
        self.runner.terminate()

    def precmd(self, line):
        """ This method is called after the line has been input but before
            it has been interpreted. If you want to modifdy the input line
            before execution (for example, variable substitution) do it here.
        """
        self._hist += [ line.strip() ]
        return line

    def postcmd(self, stop, line):
        """If you want to stop the console, return something that evaluates to true.
           If you want to do some post command processing, do it here.
        """
        return stop

    def emptyline(self):
        """Do nothing on empty input line"""
        pass

    def default(self, line):
        """Called on an input line when the command prefix is not recognized.
           In that case we execute the line as Python code.
        """
        try:
            exec(line) in self._locals, self._globals
        except Exception as e:
            print(e.__class__, ":", e)

if __name__ == '__main__':
    console = Console()
    console . cmdloop()
<commit_msg>demo of starting a plot from terminal<commit_after>import os
import sys
import cmd
import argparse
import runner
import topics
import pytelemetry.pytelemetry as tm
import pytelemetry.transports.serialtransport as transports
from ui import Plot
import numpy as np
from pyqtgraph.Qt import QtCore

class Console(cmd.Cmd):

    def __init__(self):
        cmd.Cmd.__init__(self)

        self.transport = transports.SerialTransport()
        self.telemetry = tm.pytelemetry(self.transport)
        self.topics = topics.Topics(self.telemetry)
        self.runner = runner.Runner(self.transport,self.telemetry)

        self.completekey = None
        self.prompt = ">> "
        self.intro  = "pytelemetry terminal started."  ## defaults to None

    ## Command definitions ##
    def do_ls(self, args):
        """
            Prints a list of received topics
        """
        t = self.topics.ls()
        for i in t:
            print(i)

    def do_print(self, args):
        """
            Prints X last received samples on a given topic
        """

        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        parser.add_argument('--amount','-a', dest='amount', type=int, default=10)
        args = parser.parse_args(args.split())

        s = self.topics.samples(args.topic,args.amount)
        if s is not None:
            for i in s:
                print(i)

    def do_plot(self, args):
        """
            Plots a topic in a PyQtGraph plot
        """
        if not self.topics.exists(args):
            print("Topic [",args,"] unknown.")
            return

        print("Plotting:", args)

        plt = Plot.Plot2D()

        def update(plt,topics,topic):
            data = topics.samples(topic,amount=0)
            t = np.arange(0,len(data))
            plt.trace(topic,t,data)

        timer = QtCore.QTimer()
        timer.timeout.connect(lambda: update(plt,self.topics,args))
        timer.start(50)

        data = self.topics.samples(args,amount=0)
        t = np.arange(0,len(data))
        plt.trace(args,t,data)
        plt.start()

    def do_count(self, args):
        """Counts amount of received samples on a given topic"""

        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        args = parser.parse_args(args.split())

        s = self.topics.samples(args.topic,args.amount)
        for i in s:
            print(i)

    def do_exit(self, args):
        """Exits from the terminal"""
        return -1

    def do_connect(self,args):
        """Connect to a serial port"""
        # Parse args
        parser = argparse.ArgumentParser()
        parser.add_argument('--port','-p', dest='port', required=True)
        parser.add_argument('--baudrate','-b', dest='baudrate', type=int, default=115200)
        args = parser.parse_args(args.split())

        # Build the option dictionnary
        options = dict()
        options['port'] = args.port
        options['baudrate'] = args.baudrate

        try:
            self.runner.connect(options)
        except:
            print("Failed to connect to :",options['port']," at ",options['baudrate']," (bauds)")
            print("Connection error : ",sys.exc_info())
        else:
            print("Connected to :",options['port']," at ",options['baudrate']," (bauds)")

    def do_disconnect(self,args):
        self.runner.disconnect()
        print("Disconnected.")

    def do_pub(self,args):
        """Publish data on a topic"""
        # Parse args
        parser = argparse.ArgumentParser()
        parser.add_argument('--topic','-t', dest='topic', required=True)
        parser.add_argument('--data','-d', dest='data', required=True)
        parser.add_argument('--type', dest='type', required=True)

        try:
            args = parser.parse_args(args.split())
        except Exception as e:
            print("Error :",e)
            return

        valid_types = ['string','uint8','uint16','uint32','int8','int16','int32','float32']

        if not args.type in valid_types:
            print("Invalid type. Not in ",valid_types)
            return

        if args.type == 'float32':
            args.data = float(args.data)
        elif args.type != 'string':
            args.data = int(args.data)

        print("Published on |",args.topic,"|",args.data,"[",args.type,"]")

        self.telemetry.publish(args.topic,args.data,args.type)

    ## Command definitions to support Cmd object functionality ##
    def do_help(self, args):
        """Get help on commands
           'help' or '?' with no arguments prints a list of commands for which help is available
           'help <command>' or '? <command>' gives help on <command>
        """
        ## The only reason to define this method is for the help text in the doc string
        cmd.Cmd.do_help(self, args)

    ## Override methods in Cmd object ##
    def preloop(self):
        """Initialization before prompting user for commands.
           Despite the claims in the Cmd documentaion, Cmd.preloop() is not a stub.
        """
        cmd.Cmd.preloop(self)   ## sets up command completion
        self._hist    = []      ## No history yet
        self._locals  = {}      ## Initialize execution namespace for user
        self._globals = {}

    def postloop(self):
        """Take care of any unfinished business.
           Despite the claims in the Cmd documentaion, Cmd.postloop() is not a stub.
        """
        cmd.Cmd.postloop(self)   ## Clean up command completion
        self.runner.terminate()

    def precmd(self, line):
        """ This method is called after the line has been input but before
            it has been interpreted. If you want to modifdy the input line
            before execution (for example, variable substitution) do it here.
        """
        self._hist += [ line.strip() ]
        return line

    def postcmd(self, stop, line):
        """If you want to stop the console, return something that evaluates to true.
           If you want to do some post command processing, do it here.
        """
        return stop

    def emptyline(self):
        """Do nothing on empty input line"""
        pass

    def default(self, line):
        """Called on an input line when the command prefix is not recognized.
           In that case we execute the line as Python code.
        """
        try:
            exec(line) in self._locals, self._globals
        except Exception as e:
            print(e.__class__, ":", e)

if __name__ == '__main__':
    console = Console()
    console . cmdloop()

{'fraction_non_alphanumeric': 0.12224788298691301, 'fraction_numerical': 0.004695919938414165, 'mean_word_length': 2.179393049437102, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2721817', 'n_tokens_mistral': 3721, 'n_tokens_neox': 3616, 'n_words': 1073}