Python SDK: Stream Demo Cleanup (#788)

* Clean up console for preview stream
* Fix Macos 15 wifi

Author: tcamise-gpsw

demos/python/sdk_wireless_camera_control/open_gopro/demos/gui/preview_stream.py

@@ -5,14 +5,17 @@
 
 import argparse
 import asyncio
+import logging
 
+from returns.pipeline import is_successful
 from rich.console import Console
 
 from open_gopro import WirelessGoPro
 from open_gopro.demos.gui.video_display import BufferlessVideoCapture
 from open_gopro.models import streaming
 from open_gopro.util import add_cli_args_and_parse
-from open_gopro.util.logger import setup_logging
+from open_gopro.util.logger import set_stream_logging_level, setup_logging
+from open_gopro.util.util import ainput
 
 console = Console()
 
@@ -21,22 +24,36 @@ async def main(args: argparse.Namespace) -> None:
     setup_logging(__name__, args.log)
 
     async with WirelessGoPro(args.identifier) as gopro:
-        console.print("Starting preview stream...")
-        await gopro.streaming.start_stream(
-            streaming.StreamType.PREVIEW,
-            streaming.PreviewStreamOptions(port=args.port),
+        set_stream_logging_level(logging.WARNING)
+
+        with console.status("Starting preview stream..."):
+            if not is_successful(
+                result := (
+                    await gopro.streaming.start_stream(
+                        stream_type=streaming.StreamType.PREVIEW,
+                        options=streaming.PreviewStreamOptions(port=args.port),
+                    )
+                )
+            ):
+                console.print(f"[red]Failed to start preview stream: {result.failure()}")
+                return
+            assert gopro.streaming.url, "Preview stream URL should not be empty after starting the stream"
+
+        console.print("Preview stream started. It can be viewed in VLC at [yellow]udp://@:8554")
+        console.print(
+            "Press Enter to view the preview stream using Python CV2. Once started, it won't be viewable in VLC."
         )
+        await ainput("")
 
-        console.print("Displaying the preview stream...")
-        assert gopro.streaming.url
-        BufferlessVideoCapture(
-            source=gopro.streaming.url,
-            protocol=BufferlessVideoCapture.Protocol.TS,
-            printer=console.print,
-        ).display_blocking()
+        with console.status("Displaying the preview stream..."):
+            BufferlessVideoCapture(
+                source=gopro.streaming.url,
+                protocol=BufferlessVideoCapture.Protocol.TS,
+                printer=console.print,
+            ).display_blocking()
 
-        console.print("Stopping preview stream...")
-        await gopro.streaming.stop_active_stream()
+        with console.status("Stopping preview stream..."):
+            await gopro.streaming.stop_active_stream()
 
 
 def parse_arguments() -> argparse.Namespace:

demos/python/sdk_wireless_camera_control/open_gopro/network/wifi/adapters/wireless.py

@@ -185,11 +185,11 @@ async def disconnect(self) -> bool:  # type: ignore
         """
 
     @pass_through_to_driver
-    def current(self) -> tuple[Optional[str], SsidState]:  # type: ignore
+    def current(self) -> tuple[str | None, SsidState]:  # type: ignore
         """Return the SSID and state of the current network.
 
         Returns:
-            tuple[Optional[str], SsidState]: Tuple of SSID str and state. If SSID is None,
+            tuple[str | None, SsidState]: Tuple of SSID str and state. If SSID is None,
             there is no current connection.
         """
 
@@ -222,13 +222,13 @@ def interface(self) -> str:
         return self._driver.interface
 
     @interface.setter
-    def interface(self, interface: Optional[str]) -> None:
+    def interface(self, interface: str | None) -> None:
         """Set the Wifi interface.
 
         If None is passed, interface will attempt to be auto-detected
 
         Args:
-            interface (Optional[str]): interface (or None)
+            interface (str | None): interface (or None)
         """
         self._driver.interface = interface  # type: ignore
 
@@ -245,7 +245,7 @@ def is_on(self) -> bool:
 class NmcliWireless(WifiController):
     """Linux nmcli Driver < 0.9.9.0."""
 
-    def __init__(self, password: str | None, interface: Optional[str] = None) -> None:
+    def __init__(self, password: str | None, interface: str | None = None) -> None:
         WifiController.__init__(self, interface=interface, password=password)
 
     def _clean(self, partial: str) -> None:
@@ -350,11 +350,11 @@ async def disconnect(self) -> bool:
         """
         return False
 
-    def current(self) -> tuple[Optional[str], SsidState]:
+    def current(self) -> tuple[str | None, SsidState]:
         """[summary].
 
         Returns:
-            tuple[Optional[str], SsidState]: [description]
+            tuple[str | None, SsidState]: [description]
         """
         # list active connections for all interfaces
         response = cmd(f'{self.sudo} nmcli con status | grep "{self.interface}"')
@@ -415,7 +415,7 @@ def power(self, power: bool) -> bool:
 class Nmcli0990Wireless(WifiController):
     """Linux nmcli Driver >= 0.9.9.0."""
 
-    def __init__(self, password: str | None, interface: Optional[str] = None) -> None:
+    def __init__(self, password: str | None, interface: str | None = None) -> None:
         WifiController.__init__(self, interface=interface, password=password)
 
     def _clean(self, partial: str) -> None:
@@ -516,11 +516,11 @@ async def disconnect(self) -> bool:
         """
         return False
 
-    def current(self) -> tuple[Optional[str], SsidState]:
+    def current(self) -> tuple[str | None, SsidState]:
         """[summary].
 
         Returns:
-            tuple[Optional[str], SsidState]: [description]
+            tuple[str | None, SsidState]: [description]
         """
         # list active connections for all interfaces
         response = cmd(f'{self.sudo} nmcli con | grep "{self.interface}"')
@@ -583,7 +583,7 @@ class WpasupplicantWireless(WifiController):
 
     _file = "/tmp/wpa_supplicant.conf"
 
-    def __init__(self, password: str | None, interface: Optional[str] = None) -> None:
+    def __init__(self, password: str | None, interface: str | None = None) -> None:
         WifiController.__init__(self, interface=interface, password=password)
 
     async def connect(self, ssid: str, password: str, timeout: float = 15) -> bool:
@@ -641,11 +641,11 @@ async def disconnect(self) -> bool:
         """
         return False
 
-    def current(self) -> tuple[Optional[str], SsidState]:
+    def current(self) -> tuple[str | None, SsidState]:
         """[summary].
 
         Returns:
-            tuple[Optional[str], SsidState]: [description]
+            tuple[str | None, SsidState]: [description]
         """
         # get interface status
         response = cmd(f'{self.sudo} iwconfig "{self.interface}"')
@@ -705,7 +705,7 @@ def power(self, power: bool) -> bool:
 class NetworksetupWireless(WifiController):
     """OS X networksetup Driver."""
 
-    def __init__(self, interface: Optional[str] = None) -> None:
+    def __init__(self, interface: str | None = None) -> None:
         WifiController.__init__(self, interface)
 
     async def connect(self, ssid: str, password: str, timeout: float = 15) -> bool:
@@ -779,20 +779,26 @@ async def disconnect(self) -> bool:
         """
         return False
 
-    def current(self) -> tuple[Optional[str], SsidState]:
-        """[summary].
+    def current(self) -> tuple[str | None, SsidState]:
+        """Get the currently connected SSID if there is one.
 
         Returns:
-            tuple[Optional[str], SsidState]: [description]
+            tuple[str | None, SsidState]: (SSID or None if not connected, SSID state)
         """
         # attempt to get current network
-        response = cmd(f"networksetup -getairportnetwork '{self.interface}'")
-
-        # parse response
-        phrase = "Current Wi-Fi Network: "
-        if phrase in response:
-            return (response.replace("Current Wi-Fi Network: ", "").strip(), SsidState.CONNECTED)
-        return (None, SsidState.DISCONNECTED)
+        ssid: str | None = None
+        # On MacOS <= 14...
+        if int(platform.mac_ver()[0].split(".")[0]) <= 14:
+            try:
+                if "Current Wi-Fi Network: " in (output := cmd(f"networksetup -getairportnetwork {self.interface}")):
+                    ssid = output.replace("Current Wi-Fi Network: ", "").strip()
+            except Exception as e:
+                pass
+        # For current MacOs versions or if above failed
+        if match := re.search(r"\n\s+SSID : ([\x20-\x7E]{1,32})", cmd(f"ipconfig getsummary {self.interface}")):
+            ssid = match.group(1)
+
+        return (ssid, SsidState.CONNECTED) if ssid else (None, SsidState.DISCONNECTED)
 
     def available_interfaces(self) -> list[str]:
         """Return a list of available Wifi Interface strings
@@ -875,9 +881,9 @@ class NetshWireless(WifiController):
     </MacRandomization>
 </WLANProfile>"""
 
-    def __init__(self, interface: Optional[str] = None) -> None:
+    def __init__(self, interface: str | None = None) -> None:
         WifiController.__init__(self, interface)
-        self.ssid: Optional[str] = None
+        self.ssid: str | None = None
 
     def __del__(self) -> None:
         self._clean(self.ssid)
@@ -957,7 +963,7 @@ async def disconnect(self) -> bool:
 
         return bool("completed successfully" in response.lower())
 
-    def current(self) -> tuple[Optional[str], SsidState]:
+    def current(self) -> tuple[str | None, SsidState]:
         """Get the current network SSID and state.
 
         # Here is an example of what we are parsing (i.e. to find FunHouse SSID):
@@ -969,7 +975,7 @@ def current(self) -> tuple[Optional[str], SsidState]:
         # SSID                   : FunHouse
 
         Returns:
-            tuple[Optional[str], SsidState]: Tuple of (ssid, network_state)
+            tuple[str | None, SsidState]: Tuple of (ssid, network_state)
         """
 
         class ParseState(Enum):
@@ -981,8 +987,8 @@ class ParseState(Enum):
 
         response = cmd("netsh wlan show interfaces")
         parse_state = ParseState.PARSE_INTERFACE
-        ssid: Optional[str] = None
-        network_state: Optional[str] = None
+        ssid: str | None = None
+        network_state: str | None = None
         for field in response.split("\r\n"):
             if parse_state is ParseState.PARSE_INTERFACE:
                 if "Name" in field and self.interface in field:
@@ -1053,11 +1059,11 @@ def power(self, power: bool) -> bool:
         return "not exist" not in response
 
     @staticmethod
-    def _clean(ssid: Optional[str]) -> None:
+    def _clean(ssid: str | None) -> None:
         """Disconnect and delete SSID profile.
 
         Args:
-            ssid (Optional[str]): name of SSID
+            ssid (str | None): name of SSID
         """
         cmd("netsh wlan disconnect")
         if ssid is not None: