Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Always debug log to file #453

Merged
merged 6 commits into from
Dec 7, 2023
Merged

Always debug log to file #453

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4d45294
cleanup
tcamise-gpsw Dec 7, 2023
f81d34c
Always log bluetoothctl communication to file
tcamise-gpsw Dec 7, 2023
3bb2f74
cleanup
tcamise-gpsw Dec 7, 2023
cc7f02e
Always log bluetoothctl communication to file
tcamise-gpsw Dec 7, 2023
f35dd1f
Merge branch 'python_sdk_maintenance' of github.com:gopro/OpenGoPro i…
tcamise-gpsw Dec 7, 2023
a4d8585
Fix linting
tcamise-gpsw Dec 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 36 additions & 31 deletions demos/python/sdk_wireless_camera_control/open_gopro/ble/adapters/bleak_wrapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
import asyncio
import logging
import platform
import sys
import tempfile
from pathlib import Path
from typing import Any, Callable, Optional, Pattern

import bleak
Expand Down Expand Up @@ -254,37 +255,41 @@ async def pair(self, handle: bleak.BleakClient) -> None:
handle (bleak.BleakClient): Device to pair to
"""
logger.debug("Attempting to pair...")

if (OS := platform.system()) == "Linux":
logger.info("Pairing with bluetoothctl")
# Manually control bluetoothctl on Linux
bluetoothctl = pexpect.spawn("bluetoothctl")
if logger.level == logging.DEBUG:
bluetoothctl.logfile = sys.stdout.buffer
bluetoothctl.expect("Agent registered")
# Get the version
bluetoothctl.sendline("version")
bluetoothctl.expect(r"Version")
bluetoothctl.expect(r"\n")
version = Version(bluetoothctl.before.decode("utf-8").strip())
# First see if we are already paired
if version >= Version("5.66"):
bluetoothctl.sendline("devices Paired")
bluetoothctl.expect("devices Paired")
else:
bluetoothctl.sendline("paired-devices")
bluetoothctl.expect("paired-devices")
bluetoothctl.expect(r"#")
for device in bluetoothctl.before.decode("utf-8").splitlines():
if "Device" in device and device.split()[1] == handle.address:
break # The device is already paired
else:
# We're not paired so do it now
bluetoothctl.sendline(f"pair {handle.address}")
if (match := bluetoothctl.expect(["Accept pairing", "Pairing successful"])) == 0:
bluetoothctl.sendline("yes")
bluetoothctl.expect("Pairing successful")
elif match == 1: # We received pairing successful so nothing else to do
pass
temp_file = Path(tempfile.gettempdir()) / "pexpect.log"
with open(temp_file, "wb") as fp:
logger.info("Pairing with bluetoothctl")
# Manually control bluetoothctl on Linux
bluetoothctl = pexpect.spawn("bluetoothctl")
bluetoothctl.logfile = fp
bluetoothctl.expect("Agent registered")
# Get the version
bluetoothctl.sendline("version")
bluetoothctl.expect(r"Version")
bluetoothctl.expect(r"\n")
version = Version(bluetoothctl.before.decode("utf-8").strip())
# First see if we are already paired
if version >= Version("5.66"):
bluetoothctl.sendline("devices Paired")
bluetoothctl.expect("devices Paired")
else:
bluetoothctl.sendline("paired-devices")
bluetoothctl.expect("paired-devices")
bluetoothctl.expect(r"#")
for device in bluetoothctl.before.decode("utf-8").splitlines():
if "Device" in device and device.split()[1] == handle.address:
break # The device is already paired
else:
# We're not paired so do it now
bluetoothctl.sendline(f"pair {handle.address}")
if (match := bluetoothctl.expect(["Accept pairing", "Pairing successful"])) == 0:
bluetoothctl.sendline("yes")
bluetoothctl.expect("Pairing successful")
elif match == 1: # We received pairing successful so nothing else to do
pass

logger.debug(temp_file.read_bytes().decode("utf-8"))

elif OS == "Darwin":
# No pairing on Mac
Expand Down
4 changes: 2 additions & 2 deletions demos/python/sdk_wireless_camera_control/open_gopro/demos/gui/livestream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ async def wait_for_livestream_start(_: Any, update: proto.NotifyLiveStreamStatus
console.print("[yellow]Starting livestream")
assert (await gopro.ble_command.set_shutter(shutter=Params.Toggle.ENABLE)).ok

console.print("Livestream is now streaming and should be available for viewing.")
console.print("[yellow]Livestream is now streaming and should be available for viewing.")
await ainput("Press enter to stop livestreaming...\n")

await gopro.ble_command.set_shutter(shutter=Params.Toggle.DISABLE)
Expand All @@ -80,7 +80,7 @@ def parse_arguments() -> argparse.Namespace:
default=proto.EnumWindowSize.WINDOW_SIZE_720,
)
parser.add_argument(
"--fov", help="Field of View.", choices=list(proto.EnumLens.values()), default=proto.EnumLens.LENS_WIDE
"--fov", help="Field of View.", choices=list(proto.EnumLens.values()), default=proto.EnumLens.LENS_LINEAR
)
return add_cli_args_and_parse(parser, wifi=False)

Expand Down
49 changes: 25 additions & 24 deletions demos/python/sdk_wireless_camera_control/open_gopro/logger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,29 +39,29 @@ def __init__(
self,
logger: Any,
output: Path | None = None,
modules: dict[str, int] | None = None,
modules: list[str] | None = None,
) -> None:
self.modules = {
"open_gopro.gopro_base": logging.DEBUG, # TRACE for raw HTTP responses
"open_gopro.gopro_wired": logging.DEBUG, # TRACE for concurrency debugging
"open_gopro.gopro_wireless": logging.DEBUG, # TRACE for concurrency debugging
"open_gopro.api.builders": logging.DEBUG,
"open_gopro.api.http_commands": logging.DEBUG,
"open_gopro.api.ble_commands": logging.DEBUG,
"open_gopro.communication_client": logging.DEBUG,
"open_gopro.ble.adapters.bleak_wrapper": logging.INFO, # DEBUG for pexpect communication
"open_gopro.ble.client": logging.DEBUG,
"open_gopro.wifi.adapters.wireless": logging.DEBUG,
"open_gopro.wifi.mdns_scanner": logging.DEBUG,
"open_gopro.responses": logging.DEBUG,
"open_gopro.util": logging.DEBUG,
"bleak": logging.DEBUG,
"urllib3": logging.DEBUG,
"http.client": logging.DEBUG,
}
self.modules = [
"open_gopro.gopro_base",
"open_gopro.gopro_wired",
"open_gopro.gopro_wireless",
"open_gopro.api.builders",
"open_gopro.api.http_commands",
"open_gopro.api.ble_commands",
"open_gopro.communication_client",
"open_gopro.ble.adapters.bleak_wrapper",
"open_gopro.ble.client",
"open_gopro.wifi.adapters.wireless",
"open_gopro.wifi.mdns_scanner",
"open_gopro.responses",
"open_gopro.util",
"bleak",
"urllib3",
"http.client",
]

self.logger = logger
self.modules = {**self.modules, **modules} if modules else self.modules
self.modules = modules or self.modules
self.handlers: list[logging.Handler] = []

# monkey-patch a `print` global into the http.client module; all calls to
Expand All @@ -78,7 +78,8 @@ def __init__(
datefmt="%H:%M:%S",
)
self.file_handler.setFormatter(file_formatter)
self.file_handler.setLevel(logging.TRACE) # type: ignore # pylint: disable=no-member
# Set to TRACE for concurrency debugging
self.file_handler.setLevel(logging.DEBUG)
logger.addHandler(self.file_handler)
self.addLoggingHandler(self.file_handler)
else:
Expand Down Expand Up @@ -121,9 +122,9 @@ def addLoggingHandler(self, handler: logging.Handler) -> None:
self.handlers.append(handler)

# Enable / disable logging in modules
for module, level in self.modules.items():
for module in self.modules:
l = logging.getLogger(module)
l.setLevel(level)
l.setLevel(logging.TRACE) # type: ignore
l.addHandler(handler)

# From https://stackoverflow.com/questions/2183233/how-to-add-a-custom-loglevel-to-pythons-logging-facility/35804945#35804945
Expand Down Expand Up @@ -203,7 +204,7 @@ def build_log_rx_str(stringable: Any, asynchronous: bool = False) -> str:


def setup_logging(
base: logging.Logger | str, output: Path | None = None, modules: dict[str, int] | None = None
base: logging.Logger | str, output: Path | None = None, modules: list[str] | None = None
) -> logging.Logger:
"""Configure the GoPro modules for logging and get a logger that can be used by the application
Expand Down
Loading