fixed AppendDestination, still need to check if it makes a double

Author: teto

mptcpanalyzer/cli.py

@@ -529,6 +529,9 @@ def do_tcp_summary(self, args, unknown):
         # print(args)
 
         for dest in ConnectionRoles:
+            # TODO do it only when needed
+            # con = TcpConnection.build_from_dataframe(df, tcpstreamid)
+            # df2 = tcpdest_from_connections(df, con)
             res = tcp_get_stats(
                 self.data, args.tcpstream,
                 dest,
@@ -737,8 +740,9 @@ def do_summary_extended(self, args, unknown):
                 # TODO print subflow
                 msg = inspect.cleandoc("""
                 tcpstream {sf.tcpstreamid} analysis:
-                - throughput: transferred {sf.throughput_bytes} out of {mptcp.mptcp_throughput_bytes}, accounting for {mptcp_tput_ratio:.2f}%
-                - goodput: transferred {sf.mptcp_goodput_bytes} out of {mptcp.mptcp_goodput_bytes}, accounting for {mptcp_gput_ratio:.2f}%""")
+                - throughput: transferred {sf.throughput_bytes} out of {mptcp.mptcp_throughput_bytes} mptcp bytes, accounting for {mptcp_tput_ratio:.2f}% of MPTCP throughput
+                - goodput: transferred {sf.mptcp_goodput_bytes} out of {mptcp.mptcp_goodput_bytes}, accounting for {mptcp_gput_ratio:.2f}% of MPTCP goodput
+                """)
 
                 # print(subflow)
                 self.poutput(
@@ -897,6 +901,12 @@ def do_qualify_reinjections(self, args, unknown):
 
         TODO move the code into a proper function
         """
+
+        print("Qualifying reinjections for stream in destination:")
+        # destinations = [ ConnectionRoles.Server ]
+        destinations = args.pcap_destinations
+        print("Looking at destinations %s" % destinations)
+
         # TODO this should be done automatically right ?
         df_all = load_merged_streams_into_pandas(
             args.pcap1,
@@ -925,10 +935,10 @@ def _print_reinjection_comparison(original_packet, reinj, ):
             # original_packet  = sender_df.loc[ sender_df.packetid == initial_packetid, ].iloc[0]
             row = reinj
 
-            reinjection_packetid = getattr(row, _sender("packetid")),
-            reinjection_start    = getattr(row, _sender("abstime")),
-            reinjection_arrival  = getattr(row, _receiver("abstime")),
-            original_start       = original_packet[_sender("abstime")],
+            reinjection_packetid = getattr(row, _sender("packetid"))
+            reinjection_start    = getattr(row, _sender("abstime"))
+            reinjection_arrival  = getattr(row, _receiver("abstime"))
+            original_start       = original_packet[_sender("abstime")]
             original_arrival     = original_packet[_receiver("abstime")]
 
             if reinj.redundant == False:
@@ -983,15 +993,14 @@ def _print_reinjection_comparison(original_packet, reinj, ):
             )
             return
 
-        for destination in ConnectionRoles:
-
-            if args.destinations and destination not in args.destinations:
-                log.debug("ignoring destination %s " % destination)
-                continue
+        # TODO  use args.mptcp_destinations instead
+        # TODO revert
+        # destinations = [ ConnectionRoles.Server ]
+        for destination in destinations:
 
             self.poutput("looking for reinjections towards mptcp %s" % destination)
             sender_df = df[df.mptcpdest == destination]
-            log.debug("%d reinjections in that direction" % (len(sender_df), ))
+            log.debug("%d packets in that direction" % (len(sender_df), ))
 
             # TODO we now need to display successful reinjections
             reinjections = sender_df[pd.notnull(sender_df[_sender("reinjection_of")])]
@@ -1019,6 +1028,7 @@ def _print_reinjection_comparison(original_packet, reinj, ):
     parser = MpTcpAnalyzerParser(
         description="Listing reinjections of the connection"
     )
+    # action= filter_stream
     parser.add_argument("mptcpstream", type=MpTcpStreamId, help="mptcp.stream id")
     parser.add_argument("--summary", action="store_true", default=False,
             help="Just count reinjections")

mptcpanalyzer/data.py

@@ -285,7 +285,6 @@ def _gen_converters() -> Dict[str, Callable]:
                 for name, converter in default_converters.items():
                     converters.update({_first(name): converter, _second(name): converter})
 
-
                 return converters
 
             with open(cachename) as fd:
@@ -310,6 +309,14 @@ def _gen_converters() -> Dict[str, Callable]:
                 # at this stage, destinatiosn are nan
                 # debug_dataframe(merged_df, "Merged dataframe", )
 
+                # workaround bug https://github.com/pandas-dev/pandas/issues/25448
+                def _convert_to_enums():
+                    # per_pcap_artificial_fields
+                    for col in [ _first("tcpdest"), _first("mptcpdest"), _second("tcpdest"), _second("mptcpdest")]:
+                        merged_df[col] = merged_df[col].apply(_convert_role, convert_dtype=False)
+
+                # df[]
+
                 # log.debug("Column names after loading from cache: %s", merged_df.columns)
 
         # we fix the clocks a posteriori so that the cache is still usable
@@ -638,7 +645,7 @@ def _rename_column(col_name, suffixes) -> str:
         # subdf[ _first("tcpdest") == ConnectionRole.Client] .rename(columns=_rename_cols, inplace=True)
         # print(subdf.columns)
         # print(total.columns)
-    debug_dataframe(total, "total")
+    # debug_dataframe(total, "total")
 
     logging.debug("Converted to sender/receiver format")
     return total
@@ -1173,6 +1180,7 @@ def classify_reinjections(df_all: pd.DataFrame) -> pd.DataFrame:
     Returns
         a new dataframe with an added column "redundant"
     """
+    log.debug("Classifying reinjections")
 
     df_all["redundant"] = False
     df_all["reinj_delta"] = np.nan
@@ -1187,6 +1195,7 @@ def classify_reinjections(df_all: pd.DataFrame) -> pd.DataFrame:
 
     for destination in ConnectionRoles:
 
+        log.debug("Looking at mptcp destination %r" % destination)
         sender_df = df[df.mptcpdest == destination]
 
         # print(sender_df[ sender_df.reinjected_in.notna() ][["packetid", "reinjected_in"]])
@@ -1198,7 +1207,7 @@ def classify_reinjections(df_all: pd.DataFrame) -> pd.DataFrame:
         # print(sender_df["reinjection_of"])
         reinjected_packets = sender_df.dropna(axis='index', subset=[_sender("reinjection_of")])
 
-        logging.debug("%d reinjected packets" % len(reinjected_packets))
+        log.debug("%d reinjected packets" % len(reinjected_packets))
         # with pd.option_context('display.max_rows', None, 'display.max_columns', 300):
         #     print(reinjected_packets[
         #         _sender(["packetid", "reinjected_in", "reinjection_of"]) + _receiver(["reinjected_in", "reinjection_of"])
@@ -1230,7 +1239,7 @@ def classify_reinjections(df_all: pd.DataFrame) -> pd.DataFrame:
 
             if original_packet.merge_status != "both":
                 # TODO count missed classifications ?
-                logging.debug("Original packet %d could not be mapped, giving up..." % (original_packet.packetid))
+                log.debug("Original packet %d could not be mapped, giving up..." % (original_packet.packetid))
                 continue
 
             orig_arrival = getattr(original_packet, _receiver("reltime"))

mptcpanalyzer/parser.py

@@ -180,13 +180,13 @@ class AppendDestination(DataframeAction):
     assume convention on naming
     TODO check if it's ok with FilterDest
     """
-
     def __init__(self, *args, **kwargs) -> None:
         self.already_called = False
-        # self.destinations = list(ConnectionRoles)
+        self.destinations = list(ConnectionRoles)
                     # default=list(ConnectionRoles),
         super().__init__(
-            *args, choices=CustomConnectionRolesChoices([e.name for e in ConnectionRoles]),
+            *args,
+            choices=CustomConnectionRolesChoices([e.name for e in ConnectionRoles]),
             default = list(ConnectionRoles), **kwargs)
 
 
@@ -198,11 +198,17 @@ def __call__(self, parser, namespace, values, option_string=None):
             # TODO change the default ?
             # setattr(namespace, self.dest, [])
             print("Already set")
+            print("setting value %r" % values)
+            # print("setting value %r" % self.destinations)
             # to make it unique
-            self.destinations= list(set(self.destinations.append(values)))
+            self.destinations.append(values)
+            self.destinations= list(set(self.destinations))
+            print("new result %r" % self.destinations)
         else:
-            self.destinations = values
+            print("Received first value %s" % values)
+            self.destinations = [values]
 
+        self.already_called = True
         # df_name + "destinations"
         setattr(namespace, self.dest, self.destinations)
         # pcap1 = getattr(namespace, self.df_name + "1")
@@ -332,11 +338,11 @@ def __init__(self, df_name: str, **kwargs) -> None:
 
         # assert self.field == "tcpdest" or self.field == "mptcpdest"
         # self.mptcp = mptcp
-        # self.seen 
+        # self.seen
         # init with all destinations
         self.destinations = list(ConnectionRoles)
         self.already_called = False
-        # TODO it could set choices automatically 
+        # TODO it could set choices automatically
         super().__init__(df_name, **kwargs)
 
     def __call__(self, parser, namespace, values, option_string=None):
@@ -454,6 +460,41 @@ def gen_bicap_parser(protocol, dest=False):
     # protocol=protocol,
     return gen_pcap_parser(input_pcaps=input_pcaps, direction=dest)
 
+# argparse_completer.ACArgumentParser
+class MpTcpAnalyzerParser(argparse_completer.ACArgumentParser):
+    '''
+    Wrapper around cmd2 argparse completer
+    Should allow to switch backends easily. 
+    Also allows to do some postprocessing once the arguments are parsed
+    like loading dataframes etc
+
+    '''
+
+    # def _parse_known_args(self, arg_strings, namespace):
+    def parse_known_args(self, args=None, namespace=None):
+        """
+        override it just to postprocess arguments
+        """
+
+        # returns a 2-item tuple
+        known, unknown = super().parse_known_args(args, namespace)
+        # print("res", res)
+
+        # TODO call filter_dataframe ?
+        if getattr(known, "_dataframes", None):
+            for name, df in known._dataframes.items():
+                # print
+                # so now we can filter the destination ?
+                log.debug("dataframe [%s] in namespace" % name)
+        else:
+            log.debug("No dataframe in namespace")
+
+        # postprocessing for filtering destination
+
+        log.debug("MpTcpAnalyzerParser parser finished")
+        # TODO pass along known, dataframes ?
+        return (known, unknown)
+
 
 # map pcaps to a group
 def gen_pcap_parser(
@@ -464,7 +505,7 @@ def gen_pcap_parser(
         parents=[],
         # TODO get rid of this/skip-stream
         skip_subflows: bool = True,
-    ) -> argparse_completer.ACArgumentParser:
+        ) -> MpTcpAnalyzerParser:
         """
         Generates a parser with common options.
         This parser can be completed or overridden by its children.
@@ -543,6 +584,7 @@ def _pcap(name, pcapAction="store", filterAction="store"):
                     # choices=CustomConnectionRolesChoices([e.name for e in ConnectionRoles]),
                     # TODO check how it works/FilterDest
                     action=partial(AppendDestination, df_name),
+                    # action=partial(AppendDestination, df_name),
                     # type parameter is a function/callable
                     type=lambda x: ConnectionRoles.from_string(x),
                     help='Filter flows according to their direction'
@@ -563,32 +605,4 @@ def _pcap(name, pcapAction="store", filterAction="store"):
         return parser
 
 
-# argparse_completer.ACArgumentParser
-class MpTcpAnalyzerParser(argparse_completer.ACArgumentParser):
-    '''
-    Wrapper around cmd2 argparse completer
-    Should allow to switch backends easily. 
-    Also allows to do some postprocessing once the arguments are parsed
-    like loading dataframes etc
-
-    '''
-
-    # def _parse_known_args(self, arg_strings, namespace):
-    def parse_known_args(self, args=None, namespace=None):
-        """
-        override it just to postprocess arguments
-        """
-        res = super().parse_known_args(args, namespace)
-
-        # TODO call filter_dataframe ?
-        if getattr(res, "_dataframes", None):
-            for name, df in res._dataframes.items():
-                # print
-                # so now we can filter the destination ?
-                pass
-
-        # postprocessing for filtering destination
-
-        logging.debug("MpTcpAnalyzerParser parser finished")
-        return res