Real time analysis

src/tools/interop/idt/capture/base.py

@@ -48,7 +48,7 @@ async def start_streaming(self) -> None:
     @abstractmethod
     async def stop_streaming(self) -> None:
         """
-        Stop streaming logs
+        Stop the capture and pull any artifacts from remote devices
         Stop should not cause an error even if the stream is not running
         """
         raise NotImplementedError
@@ -84,14 +84,21 @@ async def start_capture(self) -> None:
     @abstractmethod
     async def stop_capture(self) -> None:
         """
-        Stop the capture
+        Stop the capture and pull any artifacts from remote devices
         """
         raise NotImplementedError
 
     @abstractmethod
-    async def analyze_capture(self) -> None:
+    def analyze_capture(self) -> None:
         """
         Parse the capture and create + display helpful analysis artifacts that are unique to the ecosystem
-        Write analysis artifacts to artifact_dir
+        This function will be run as a separate process to allow real time analysis
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def probe_capture(self) -> None:
+        """
+        Probe the local environment, e.g. ping relevant remote services and write respective artifacts
         """
         raise NotImplementedError

src/tools/interop/idt/capture/ecosystem/play_services/analysis.py

@@ -17,14 +17,18 @@
 
 import os
 
+from capture.utils import log
 from capture.utils.artifact import create_standard_log_name
 from capture.utils.log import print_and_write, add_border
 from capture.platform.android import Android
 
+logger = log.get_logger(__file__)
+
 
 class PlayServicesAnalysis:
 
     def __init__(self, platform: Android, artifact_dir: str) -> None:
+        self.logger = logger
         self.artifact_dir = artifact_dir
         self.analysis_file_name = os.path.join(
             self.artifact_dir, create_standard_log_name(
@@ -38,10 +42,16 @@ def __init__(self, platform: Android, artifact_dir: str) -> None:
         self.resolver_logs = ''
         self.sigma_logs = ''
         self.fail_trace_line_counter = -1
+        self.real_time = True
+
+    def rt_log(self, line):
+        if self.real_time:
+            self.logger.info(line)
 
     def _log_proc_matter_commissioner(self, line: str) -> None:
         """Core commissioning flow"""
         if 'MatterCommissioner' in line:
+            self.rt_log(line)
             self.matter_commissioner_logs += line
 
     def _log_proc_commissioning_failed(self, line: str) -> None:
@@ -52,24 +62,33 @@ def _log_proc_commissioning_failed(self, line: str) -> None:
             self.failure_stack_trace += line
             self.fail_trace_line_counter += 1
         if 'SetupDeviceView' and 'Commissioning failed' in line:
+            self.rt_log(line)
             self.fail_trace_line_counter = 0
             self.failure_stack_trace += line
 
     def _log_proc_pake(self, line: str) -> None:
         """Three logs for pake 1-3 expected"""
         if "Pake" in line and "chip_logging" in line:
+            self.rt_log(line)
             self.pake_logs += line
 
     def _log_proc_mdns(self, line: str) -> None:
         if "_matter" in line and "ServiceResolverAdapter" in line:
+            self.rt_log(line)
             self.resolver_logs += line
 
     def _log_proc_sigma(self, line: str) -> None:
         """Three logs expected for sigma 1-3"""
         if "Sigma" in line and "chip_logging" in line:
+            self.rt_log(line)
             self.sigma_logs += line
 
-    def _show_analysis(self) -> None:
+    def show_analysis(self) -> None:
+        # TODO: Because the realtime proc is forked, have to do this again for now...
+        self.real_time = False
+        with open(self.platform.streams["LogcatStreamer"].logcat_artifact, mode='r') as logcat_file:
+            for line in logcat_file:
+                self.process_line(line)
         analysis_file = open(self.analysis_file_name, mode="w+")
         print_and_write(add_border('Matter commissioner logs'), analysis_file)
         print_and_write(self.matter_commissioner_logs, analysis_file)
@@ -86,11 +105,9 @@ def _show_analysis(self) -> None:
         analysis_file.close()
 
     def process_line(self, line: str) -> None:
-        for line_func in filter(lambda s: s.startswith('_log'), dir(self)):
+        for line_func in [s for s in dir(self) if s.startswith('_log')]:
             getattr(self, line_func)(line)
 
-    def do_analysis(self) -> None:
-        with open(self.platform.streams["LogcatStreamer"].logcat_artifact, mode='r') as logcat_file:
-            for line in logcat_file:
-                self.process_line(line)
-        self._show_analysis()
+    def do_analysis(self, batch: [str]) -> None:
+        for line in batch:
+            self.process_line(line)

src/tools/interop/idt/capture/ecosystem/play_services/play_services.py

@@ -17,6 +17,7 @@
 
 import json
 import os
+import time
 from typing import Dict
 
 from capture.base import EcosystemCapture, UnsupportedCapturePlatformException
@@ -28,11 +29,11 @@
 from .command_map import dumpsys, getprop
 from .prober import PlayServicesProber
 
-
 from capture.utils import log
 
 logger = log.get_logger(__file__)
 
+
 class PlayServices(EcosystemCapture):
     """
     Implementation of capture and analysis for Play Services
@@ -97,9 +98,16 @@ async def start_capture(self) -> None:
 
     async def stop_capture(self) -> None:
         await self.platform.stop_streaming()
+        self.analysis.show_analysis()
+
+    def analyze_capture(self):
+        fd = open(self.platform.streams["LogcatStreamer"].logcat_artifact, "r")
+        while True:
+            self.analysis.do_analysis(fd.readlines())
+            time.sleep(4)
+        fd.close()
 
-    async def analyze_capture(self) -> None:
-        self.analysis.do_analysis()
+    async def probe_capture(self) -> None:
         if config.enable_foyer_probers:
             await PlayServicesProber(self.platform).probe_services()
         else:

src/tools/interop/idt/capture/ecosystem/play_services_user/play_services_user.py

@@ -46,22 +46,38 @@ async def start_capture(self) -> None:
 
     async def stop_capture(self) -> None:
         await self.platform.stop_streaming()
+        self.show_analysis()
 
-    async def analyze_capture(self) -> None:
+    @staticmethod
+    def proc_line(line, analysis_file) -> None:
+        if "CommissioningServiceBin: Binding to service" in line:
+            print_and_write(
+                f"3P commissioner initiated Play Services commissioning\n{line}",
+                analysis_file)
+        elif "CommissioningServiceBin: Sending commissioning request to bound service" in line:
+            print_and_write(
+                f"Play Services commissioning complete; passing back to 3P\n{line}",
+                analysis_file)
+        elif "CommissioningServiceBin: Received commissioning complete from bound service" in line:
+            print_and_write(
+                f"3P commissioning complete!\n{line}",
+                analysis_file)
+
+    def analyze_capture(self) -> None:
         """"Show the start and end times of commissioning boundaries"""
+        fd = open(self.platform.streams["LogcatStreamer"].logcat_artifact, "r")
+        while True:
+            for line in fd.readlines():
+                self.proc_line(line, open("/dev/null", mode="w"))
+        fd.close()
+
+    def show_analysis(self) -> None:
+        # TODO: Make more elegant
         analysis_file = open(self.analysis_file, mode='w+')
         with open(self.platform.streams["LogcatStreamer"].logcat_artifact, mode='r') as logcat_file:
             for line in logcat_file:
-                if "CommissioningServiceBin: Binding to service" in line:
-                    print_and_write(
-                        f"3P commissioner initiated Play Services commissioning\n{line}",
-                        analysis_file)
-                elif "CommissioningServiceBin: Sending commissioning request to bound service" in line:
-                    print_and_write(
-                        f"Play Services commissioning complete; passing back to 3P\n{line}",
-                        analysis_file)
-                elif "CommissioningServiceBin: Received commissioning complete from bound service" in line:
-                    print_and_write(
-                        f"3P commissioning complete!\n{line}",
-                        analysis_file)
+                self.proc_line(line, analysis_file)
         analysis_file.close()
+
+    async def probe_capture(self) -> None:
+        pass

src/tools/interop/idt/capture/factory.py

@@ -17,25 +17,27 @@
 
 import asyncio
 import copy
+import multiprocessing
 import os
 import traceback
 import typing
+from multiprocessing import Process
 
 import capture
-from capture.utils import log
 from capture.utils.async_control import get_timeout
 from capture.base import EcosystemCapture, PlatformLogStreamer, UnsupportedCapturePlatformException
 from capture.utils.artifact import safe_mkdir
 from capture.utils.log import border_print
+from capture.utils import log
 
 _PLATFORM_MAP: typing.Dict[str, PlatformLogStreamer] = {}
 _ECOSYSTEM_MAP: typing.Dict[str, PlatformLogStreamer] = {}
 _ERROR_REPORT: typing.Dict[str, list[str]] = {}
-
-from capture.utils import log
+_ANALYSIS_MAP: typing.Dict[str, Process] = {}
 
 logger = log.get_logger(__file__)
 
+
 class PlatformFactory:
 
     @staticmethod
@@ -130,12 +132,22 @@ async def start():
 
     @staticmethod
     async def stop():
+        for ecosystem_name, ecosystem in _ANALYSIS_MAP.items():
+            border_print(f"Stopping analysis proc for {ecosystem_name}")
+            ecosystem.kill()
         await EcosystemController.handle_capture("stop")
 
     @staticmethod
-    async def analyze():
-        # TODO: allow real time, not just post
-        await EcosystemController.handle_capture("analyze")
+    def analyze():
+        for ecosystem_name, ecosystem in _ECOSYSTEM_MAP.items():
+            border_print(f"Starting analyze subproc for {ecosystem_name}")
+            proc = multiprocessing.Process(target=ecosystem.analyze_capture)
+            _ANALYSIS_MAP[ecosystem_name] = proc
+            proc.start()
+
+    @staticmethod
+    async def probe():
+        await EcosystemController.handle_capture("probe")
 
     @staticmethod
     def error_report():

src/tools/interop/idt/capture/platform/android/capabilities.py

@@ -32,6 +32,7 @@ def check_snoop_log(self) -> bool:
                                                                      capture_output=True).get_captured_output()
 
     def check_capabilities(self):
+        # TODO: Write capabilities to artifacts
         self.logger.info("Checking if device has root")
         self.c_has_root = self.platform.run_adb_command(
             "shell which su", capture_output=True).finished_success()

src/tools/interop/idt/idt.py

@@ -189,18 +189,20 @@ def command_capture(self, args: argparse.Namespace) -> None:
                                                      args.ecosystem,
                                                      self.artifact_dir))
         asyncio.run(EcosystemController.start())
+        EcosystemController.analyze()
 
-        border_print("Press enter twice to stop streaming", important=True)
+        border_print("Press enter three times to stop streaming", important=True)
         input("")
 
         if pcap:
             border_print("Stopping pcap")
             pcap_runner.stop_pcap()
 
         asyncio.run(EcosystemController.stop())
-        asyncio.run(EcosystemController.analyze())
 
-        # TODO: Write error report to artifacts
+        asyncio.run(EcosystemController.probe())
+
+        # TODO: Write error traces to artifacts
         if EcosystemController.has_errors():
             border_print("Errors seen this run:")
             EcosystemController.error_report()