saving before the move

Author: teto

mptcpanalyzer/cli.py

@@ -188,6 +188,11 @@ def __init__(self, cfg: MpTcpAnalyzerConfig, stdin=sys.stdin, **kwargs) -> None:
         self.debug = True  # for now
         self.set_posix_shlex = True  # need cmd2 >= 0.8
 
+        # Pandas specific initialization
+        # for as long as https://github.com/pydata/numexpr/issues/331 is a problem
+        pd.set_option('compute.use_numexpr', False)
+        print("use numexpr?", pd.get_option('compute.use_numexpr', False))
+
         #  Load Plots
         ######################
         # you can  list available plots under the namespace
@@ -502,6 +507,7 @@ def _print_subflow(x):
     # TODO update the stream id autcompletion dynamically ?
     # setattr(action_stream, argparse_completer.ACTION_ARG_CHOICES, range(0, 10))
 
+    # TODO use filter_dest instead
     summary_parser.add_argument(
         'destination',
         # mp.DestinationChoice,
@@ -533,16 +539,12 @@ def do_summary(self, args, unknown):
         ret = mptcp_compute_throughput(
             self.data, args.mptcpstream, args.destination
         )
-        # if success is not True:
-        #     self.perror("Throughput computation failed:")
-        #     self.perror(ret)
-        #     return
 
         if args.json:
             import json
             # TODO use self.poutput
             # or use a stream, it must just be testable
-            val = json.dumps(ret, ensure_ascii=False)
+            val = json.dumps(dataclasses.asdict(ret), ensure_ascii=False)
             self.poutput(val)
             return
 

mptcpanalyzer/data.py

@@ -403,16 +403,7 @@ def load_into_pandas(
             dtypes = {field.fullname: field.type for _, field in config.fields.items() if field.converter is None}
             log.debug("Dtypes before load: %s" % dtypes)
             log.debug("Converters before load: %s" % converters)
-            # test = pd.read_csv(
-            #     fd,
-            #     comment='#',
-            #     sep=config.delimiter,
-            #     nrows=1, # useful for debugging purpose
-            # )
-            # log.debug("Dtypes after load:%s\n" % dict(test.dtypes))
-
-            # https://stackoverflow.com/questions/52686559/read-csv-get-the-line-where-exception-occured
-            # print(test.columns)
+
             from .pdutils import read_csv_debug
             fields = [f.fullname for _, f in config.fields.items()]
             # data = read_csv_debug(fields,
@@ -529,29 +520,6 @@ def mptcpdest_from_connections(df, con: MpTcpConnection) -> pd.DataFrame:
 
     return df
 
-    # for tcpdest in ConnectionRoles:
-
-    #     log.debug("Looking at tcpdestination %s" % tcpdest)
-
-    #     # pandas trick to avoid losing dtype
-    #     # see https://github.com/pandas-dev/pandas/issues/22361#issuecomment-413147667
-    #     # no need to set _second (as they are just opposite)
-    #     # TODO this should be done somewhere else
-    #     # else summary won't work
-    #     res[_first('tcpdest')][:] = tcpdest
-    #     res[_second('tcpdest')][:] = tcpdest
-
-    #     # generate_mptcp_direction_query
-    #     if isinstance(main_connection, MpTcpSubflow):
-
-    #         print("THIS IS A SUBFLOW")
-    #         mptcpdest = main_connection.mptcp_dest_from_tcpdest(tcpdest)
-    #         res[_first('mptcpdest')][:] = mptcpdest
-    #         res[_second('mptcpdest')][:] = mptcpdest
-
-    #         print("Setting mptcpdest to %s", mptcpdest)
-    #         # if tcpdest == main_connection.mptcpdest
-
 
 def tcpdest_from_connections(df, con: TcpConnection) -> pd.DataFrame:
 

mptcpanalyzer/parser.py

@@ -9,6 +9,7 @@
         tcpdest_from_connections, mptcpdest_from_connections)
 from mptcpanalyzer import (PreprocessingActions, ConnectionRoles, DestinationChoice,
             CustomConnectionRolesChoices, TcpStreamId, MpTcpStreamId)
+import mptcpanalyzer as mp
 from functools import partial
 from mptcpanalyzer.connection import MpTcpConnection, TcpConnection
 
@@ -40,6 +41,9 @@ def _add_dataframe(namespace, dest, df):
 
 
 class DataframeAction(argparse.Action):
+    '''
+    If you need the action to act on a specific dataframe
+    '''
 
     def __init__(self, df_name: str, **kwargs) -> None:
         argparse.Action.__init__(self, **kwargs)
@@ -52,10 +56,6 @@ def add_dataframe(self, namespace, df):
 
 
 
-# class StreamId(x):
-#     return int(x)
-
-
 class LoadSinglePcap(DataframeAction):
     '''
     Test action !!
@@ -75,33 +75,6 @@ def __call__(self, parser, namespace, values, option_string=None):
 
         self.add_dataframe (namespace, df)
 
-# def with_argparser_test(argparser: argparse.ArgumentParser,
-#                    preserve_quotes: bool=False) -> Callable[[argparse.Namespace], Optional[bool]]:
-#     import functools
-
-#     # noinspection PyProtectedMember
-#     def arg_decorator(func: Callable[[cmd2.Statement], Optional[bool]]):
-#         @functools.wraps(func)
-#         def cmd_wrapper(instance, cmdline):
-#             lexed_arglist = cmd2.cmd2.parse_quoted_string(cmdline, preserve_quotes)
-#             return func(instance, argparser, lexed_arglist)
-
-#         # argparser defaults the program name to sys.argv[0]
-#         # we want it to be the name of our command
-#         # argparser.prog = func.__name__[len(COMMAND_FUNC_PREFIX):]
-
-#         # If the description has not been set, then use the method docstring if one exists
-#         if argparser.description is None and func.__doc__:
-#             argparser.description = func.__doc__
-
-#         # Set the command's help text as argparser.description (which can be None)
-#         # cmd_wrapper.__doc__ = argparser.description
-
-#         # Mark this function as having an argparse ArgumentParser
-#         setattr(cmd_wrapper, 'argparser', argparser)
-
-#         return cmd_wrapper
-
 #     return arg_decorator
 # def with_argparser(argparser: argparse.ArgumentParser,
 #                    preserve_quotes: bool=False) -> Callable[[argparse.Namespace], Optional[bool]]:
@@ -153,6 +126,10 @@ def with_argparser_test(
     preserve_quotes: bool=False,
     preload_pcap: bool=False,
     ) -> Callable[[argparse.Namespace, List], Optional[bool]]:
+    """
+    Arguments:
+        preload_pcap: Use the preloaded pcap as a dataframe
+    """
     import functools
 
     # noinspection PyProtectedMember
@@ -203,7 +180,6 @@ class AppendDestination(DataframeAction):
     assume convention on naming
     """
 
-    # query
     def __init__(self, *args, **kwargs) -> None:
         self.already_called = False
         # self.destinations = list(ConnectionRoles)
@@ -254,7 +230,8 @@ class MergePcaps(DataframeAction):
     """
     assume convention on naming
     """
-    def __init__(self,
+    def __init__(
+        self,
         name: str,
         protocol: str, # mptcp or tcp ?
         loader = TsharkConfig(),
@@ -327,18 +304,13 @@ def __call__(self, parser, namespace, values, option_string=None):
 #     def __
 
 # don't need the Mptcp flag anymore
-def exclude_stream(df_name, mptcp: bool = False):
-    query = "tcpstream"
-    if mptcp:
-        query = "mp" + query
-    query = query + "!={streamid}"
+def exclude_stream(df_name):
+    query = "{field}!={streamid}"
     return partial(FilterStream, query, df_name)
 
-def retain_stream(df_name, mptcp: bool = False):
-    query = "tcpstream"
-    if mptcp:
-        query = "mp" + query
-    query = query + "=={streamid}"
+# TODO va dependre du type en fait
+def retain_stream(df_name):
+    query = "{field}=={streamid}"
     return partial(FilterStream, query, df_name)
 
 
@@ -424,7 +396,6 @@ class FilterStream(DataframeAction):
     def __init__(self, query: str, df_name: str, **kwargs) -> None:
         # self.df_name = df_name
         self.query_tpl = query
-        # self.mptcp = mptcp
         super().__init__(df_name, **kwargs)
 
     def __call__(self, parser, namespace, values, option_string=None):
@@ -437,31 +408,32 @@ def __call__(self, parser, namespace, values, option_string=None):
         # make sure result
         df = namespace._dataframes[self.df_name]
 
-        # streamid = values
-
         log.debug("Filtering stream %s" % (values))
 
         # if type(values) != list:
         #     streamids = list(values)
+        print("received values %r" % values)
 
-        # TODO build a query
-        mptcp = False
         field = "tcpstream"
-        if isinstance(values, TcpStreamId):
-            pass
-
-        elif isinstance(values, MpTcpStreamId):
-            mptcp = True
+        if isinstance(values, MpTcpStreamId):
             field = "mptcpstream"
+            print("mptcp instance type ")
+        elif isinstance(values, TcpStreamId):
+            pass
         else:
-            parser.error("Unsupported type %s" % type(values))
+            parser.error("Unsupported 'type' %s. Set it to TcpStreamId or MpTcpStreamId" % type(values))
 
         # super(argparse.Action).__call__(parser, namespace, values, option_string)
         setattr(namespace, self.dest, values)
-        query = self.query_tpl.format(streamid=values)
+        query = self.query_tpl.format(field=field, streamid=values)
 
-        log.debug("Applying query %s" % query)
-        df.query(query, inplace=True)
+        log.log(mp.TRACE, "Applying query [%s]" % query)
+        print(df.head(5))
+        print(df.dtypes)
+
+        import pandas as pd
+        print("use numexpr?", pd.get_option('compute.use_numexpr', False))
+        df.query(query, inplace=True, )
 
 
 def gen_bicap_parser(protocol, dest=False):
@@ -550,12 +522,11 @@ def _pcap(name, pcapAction="store", filterAction="store"):
                 #     help=argparse.SUPPRESS)
                 # merge_pcap.default = "TEST"
             else:
-                # print("PreprocessingActions.Merge:")
-                # TODO pas forcement
                 filterClass = FilterStream
                 _pcap(df_name, pcapAction=LoadSinglePcap,
                     filterAction=retain_stream(df_name,
-                    mptcp = bool(bitfield & PreprocessingActions.FilterMpTcpStream))
+                    # mptcp = bool(bitfield & PreprocessingActions.FilterMpTcpStream)
+                        )
                 )
 
             if bitfield & PreprocessingActions.FilterDestination or direction :
@@ -582,7 +553,7 @@ def _pcap(name, pcapAction="store", filterAction="store"):
             if skip_subflows:
                 parser.add_argument(
                     '--skip', dest=df_name + "skipped_subflows", type=TcpStreamId,
-                    action=exclude_stream(df_name, mptcp=False),
+                    action=exclude_stream(df_name,),
                     default=[],
                     help=("You can type here the tcp.stream of a subflow "
                         "not to take into account (because"
@@ -601,8 +572,6 @@ class MpTcpAnalyzerParser(argparse_completer.ACArgumentParser):
 
     '''
 
-    # def __init__():
-
     # def _parse_known_args(self, arg_strings, namespace):
     def parse_known_args(self, args=None, namespace=None):
         """

mptcpanalyzer/pdutils.py

@@ -16,6 +16,8 @@ def connection(self, streamid):
         return TcpConnection.build_from_dataframe(self._obj, streamid)
 
 
+
+# https://stackoverflow.com/questions/52686559/read-csv-get-the-line-where-exception-occured
 def read_csv_debug(fields, fd, *args, first_try=True, **kwargs):
     """
     Help debugging dataframe loading errors (with dtypes/converters)
@@ -51,17 +53,8 @@ def read_csv_debug(fields, fd, *args, first_try=True, **kwargs):
                 raise e
 
         finally:
-
             fd.seek(0)
-        # else:
-        #     data = pd.read_csv(
-        #         fd,
-        #         *args,
-        #         usecols=[ field],
-        #         **kwargs
-        #     )
-
-    # return data
+
 
 
 def filter_dataframe(

setup.py

@@ -104,7 +104,7 @@ def run(self):
     ],
     # test_suite="tests",
     cmdclass={
-    "test": RunTests,
+        "test": RunTests,
     },
     zip_safe=False,
     )