<commit_before># Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# -----------------------------------------------------------------------------
# Bumble Tool
# -----------------------------------------------------------------------------

# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
from bumble.hci import HCI_Constant
import os
import os.path
import logging
import click
from collections import OrderedDict
import colors
import re

from bumble.core import UUID, AdvertisingData, TimeoutError, BT_LE_TRANSPORT
from bumble.device import ConnectionParametersPreferences, Device, Connection, Peer
from bumble.utils import AsyncRunner
from bumble.transport import open_transport_or_link
from bumble.gatt import Characteristic
from bumble.hci import (
    HCI_LE_1M_PHY,
    HCI_LE_2M_PHY,
    HCI_LE_CODED_PHY,
)

from prompt_toolkit import Application
from prompt_toolkit.history import FileHistory
from prompt_toolkit.completion import Completer, Completion, NestedCompleter
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.formatted_text import ANSI
from prompt_toolkit.styles import Style
from prompt_toolkit.filters import Condition
from prompt_toolkit.widgets import TextArea, Frame
from prompt_toolkit.widgets.toolbars import FormattedTextToolbar
from prompt_toolkit.data_structures import Point
from prompt_toolkit.layout import (
    Layout,
    HSplit,
    Window,
    CompletionsMenu,
    Float,
    FormattedTextControl,
    FloatContainer,
    ConditionalContainer,
    Dimension
)

# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
BUMBLE_USER_DIR            = os.path.expanduser('~/.bumble')
DEFAULT_RSSI_BAR_WIDTH     = 20
DEFAULT_CONNECTION_TIMEOUT = 30.0
DISPLAY_MIN_RSSI           = -100
DISPLAY_MAX_RSSI           = -30
RSSI_MONITOR_INTERVAL      = 5.0  # Seconds


# -----------------------------------------------------------------------------
# Globals
# -----------------------------------------------------------------------------
App = None


# -----------------------------------------------------------------------------
# Utils
# -----------------------------------------------------------------------------

def le_phy_name(phy_id):
    return {
        HCI_LE_1M_PHY:    '1M',
        HCI_LE_2M_PHY:    '2M',
        HCI_LE_CODED_PHY: 'CODED'
    }.get(phy_id, HCI_Constant.le_phy_name(phy_id))


def rssi_bar(rssi):
    blocks = ['', '▏', '▎', '▍', '▌', '▋', '▊', '▉']
    bar_width = (rssi - DISPLAY_MIN_RSSI) / (DISPLAY_MAX_RSSI - DISPLAY_MIN_RSSI)
    bar_width = min(max(bar_width, 0), 1)
    bar_ticks = int(bar_width * DEFAULT_RSSI_BAR_WIDTH * 8)
    bar_blocks = ('█' * int(bar_ticks / 8)) + blocks[bar_ticks % 8]
    return f'{rssi:4} {bar_blocks}'


def parse_phys(phys):
    if phys.lower() == '*':
        return None
    else:
        phy_list = []
        elements = phys.lower().split(',')
        for element in elements:
            if element == '1m':
                phy_list.append(HCI_LE_1M_PHY)
            elif element == '2m':
                phy_list.append(HCI_LE_2M_PHY)
            elif element == 'coded':
                phy_list.append(HCI_LE_CODED_PHY)
            else:
                raise ValueError('invalid PHY name')
        return phy_list


# -----------------------------------------------------------------------------
# Console App
# -----------------------------------------------------------------------------
class ConsoleApp:
    def __init__(self):
        self.known_addresses  = set()
        self.known_attributes = []
        self.device           = None
        self.connected_peer   = None
        self.top_tab          = 'scan'
        self.monitor_rssi     = False
        self.connection_rssi  = None

        style = Style.from_dict({
            'output-field': 'bg:#000044 #ffffff',
            'input-field':  'bg:#000000 #ffffff',
            'line':         '#004400',
            'error':        'fg:ansired'
        })

        class LiveCompleter(Completer):
            def __init__(self, words):
                self.words = words

            def get_completions(self, document, complete_event):
                prefix = document.text_before_cursor.upper()
                for word in [x for x in self.words if x.upper().startswith(prefix)]:
                    yield Completion(word, start_position=-len(prefix))

        def make_completer():
            return NestedCompleter.from_nested_dict({
                'scan': {
                    'on': None,
                    'off': None
                },
                'advertise': {
                    'on': None,
                    'off': None
                },
                'rssi': {
                    'on': None,
                    'off': None
                },
                'show': {
                    'scan': None,
                    'services': None,
                    'attributes': None,
                    'log': None
                },
                'filter': {
                    'address': None,
                },
                'connect': LiveCompleter(self.known_addresses),
                'update-parameters': None,
                'encrypt': None,
                'disconnect': None,
                'discover': {
                    'services': None,
                    'attributes': None
                },
                'request-mtu': None,
                'read': LiveCompleter(self.known_attributes),
                'write': LiveCompleter(self.known_attributes),
                'subscribe': LiveCompleter(self.known_attributes),
                'unsubscribe': LiveCompleter(self.known_attributes),
                'set-phy': {
                    '1m': None,
                    '2m': None,
                    'coded': None
                },
                'set-default-phy': None,
                'quit': None,
                'exit': None
            })

        self.input_field = TextArea(
            height=1,
            prompt="> ",
            multiline=False,
            wrap_lines=False,
            completer=make_completer(),
            history=FileHistory(os.path.join(BUMBLE_USER_DIR, 'history'))
        )

        self.input_field.accept_handler = self.accept_input

        self.output_height = Dimension(min=7, max=7, weight=1)
        self.output_lines = []
        self.output = FormattedTextControl(get_cursor_position=lambda: Point(0, max(0, len(self.output_lines) - 1)))
        self.output_max_lines = 20
        self.scan_results_text = FormattedTextControl()
        self.services_text = FormattedTextControl()
        self.attributes_text = FormattedTextControl()
        self.log_text = FormattedTextControl(get_cursor_position=lambda: Point(0, max(0, len(self.log_lines) - 1)))
        self.log_height = Dimension(min=7, weight=4)
        self.log_max_lines = 100
        self.log_lines = []

        container = HSplit([
            ConditionalContainer(
                Frame(Window(self.scan_results_text), title='Scan Results'),
                filter=Condition(lambda: self.top_tab == 'scan')
            ),
            ConditionalContainer(
                Frame(Window(self.services_text), title='Services'),
                filter=Condition(lambda: self.top_tab == 'services')
            ),
            ConditionalContainer(
                Frame(Window(self.attributes_text), title='Attributes'),
                filter=Condition(lambda: self.top_tab == 'attributes')
            ),
            ConditionalContainer(
                Frame(Window(self.log_text, height=self.log_height), title='Log'),
                filter=Condition(lambda: self.top_tab == 'log')
            ),
            Frame(Window(self.output, height=self.output_height)),
            FormattedTextToolbar(text=self.get_status_bar_text, style='reverse'),
            self.input_field
        ])

        container = FloatContainer(
            container,
            floats=[
                Float(
                    xcursor=True,
                    ycursor=True,
                    content=CompletionsMenu(max_height=16, scroll_offset=1),
                ),
            ],
        )

        layout = Layout(container, focused_element=self.input_field)

        kb = KeyBindings()
        @kb.add("c-c")
        @kb.add("c-q")
        def _(event):
            event.app.exit()

        self.ui = Application(
            layout=layout,
            style=style,
            key_bindings=kb,
            full_screen=True
        )

    async def run_async(self, device_config, transport):
        rssi_monitoring_task = asyncio.create_task(self.rssi_monitor_loop())

        async with await open_transport_or_link(transport) as (hci_source, hci_sink):
            if device_config:
                self.device = Device.from_config_file_with_hci(device_config, hci_source, hci_sink)
            else:
                self.device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink)
            self.device.listener = DeviceListener(self)
            await self.device.power_on()

            # Run the UI
            await self.ui.run_async()
<commit_msg>Use random address in console.py if device config is not provided<commit_after># Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# -----------------------------------------------------------------------------
# Bumble Tool
# -----------------------------------------------------------------------------

# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import logging
import os
import random
import re
from collections import OrderedDict

import click
import colors

from bumble.core import UUID, AdvertisingData, TimeoutError, BT_LE_TRANSPORT
from bumble.device import ConnectionParametersPreferences, Device, Connection, Peer
from bumble.utils import AsyncRunner
from bumble.transport import open_transport_or_link
from bumble.gatt import Characteristic
from bumble.hci import (
    HCI_Constant,
    HCI_LE_1M_PHY,
    HCI_LE_2M_PHY,
    HCI_LE_CODED_PHY,
)

from prompt_toolkit import Application
from prompt_toolkit.history import FileHistory
from prompt_toolkit.completion import Completer, Completion, NestedCompleter
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.formatted_text import ANSI
from prompt_toolkit.styles import Style
from prompt_toolkit.filters import Condition
from prompt_toolkit.widgets import TextArea, Frame
from prompt_toolkit.widgets.toolbars import FormattedTextToolbar
from prompt_toolkit.data_structures import Point
from prompt_toolkit.layout import (
    Layout,
    HSplit,
    Window,
    CompletionsMenu,
    Float,
    FormattedTextControl,
    FloatContainer,
    ConditionalContainer,
    Dimension
)

# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
BUMBLE_USER_DIR            = os.path.expanduser('~/.bumble')
DEFAULT_RSSI_BAR_WIDTH     = 20
DEFAULT_CONNECTION_TIMEOUT = 30.0
DISPLAY_MIN_RSSI           = -100
DISPLAY_MAX_RSSI           = -30
RSSI_MONITOR_INTERVAL      = 5.0  # Seconds


# -----------------------------------------------------------------------------
# Globals
# -----------------------------------------------------------------------------
App = None


# -----------------------------------------------------------------------------
# Utils
# -----------------------------------------------------------------------------

def le_phy_name(phy_id):
    return {
        HCI_LE_1M_PHY:    '1M',
        HCI_LE_2M_PHY:    '2M',
        HCI_LE_CODED_PHY: 'CODED'
    }.get(phy_id, HCI_Constant.le_phy_name(phy_id))


def rssi_bar(rssi):
    blocks = ['', '▏', '▎', '▍', '▌', '▋', '▊', '▉']
    bar_width = (rssi - DISPLAY_MIN_RSSI) / (DISPLAY_MAX_RSSI - DISPLAY_MIN_RSSI)
    bar_width = min(max(bar_width, 0), 1)
    bar_ticks = int(bar_width * DEFAULT_RSSI_BAR_WIDTH * 8)
    bar_blocks = ('█' * int(bar_ticks / 8)) + blocks[bar_ticks % 8]
    return f'{rssi:4} {bar_blocks}'


def parse_phys(phys):
    if phys.lower() == '*':
        return None
    else:
        phy_list = []
        elements = phys.lower().split(',')
        for element in elements:
            if element == '1m':
                phy_list.append(HCI_LE_1M_PHY)
            elif element == '2m':
                phy_list.append(HCI_LE_2M_PHY)
            elif element == 'coded':
                phy_list.append(HCI_LE_CODED_PHY)
            else:
                raise ValueError('invalid PHY name')
        return phy_list


# -----------------------------------------------------------------------------
# Console App
# -----------------------------------------------------------------------------
class ConsoleApp:
    def __init__(self):
        self.known_addresses  = set()
        self.known_attributes = []
        self.device           = None
        self.connected_peer   = None
        self.top_tab          = 'scan'
        self.monitor_rssi     = False
        self.connection_rssi  = None

        style = Style.from_dict({
            'output-field': 'bg:#000044 #ffffff',
            'input-field':  'bg:#000000 #ffffff',
            'line':         '#004400',
            'error':        'fg:ansired'
        })

        class LiveCompleter(Completer):
            def __init__(self, words):
                self.words = words

            def get_completions(self, document, complete_event):
                prefix = document.text_before_cursor.upper()
                for word in [x for x in self.words if x.upper().startswith(prefix)]:
                    yield Completion(word, start_position=-len(prefix))

        def make_completer():
            return NestedCompleter.from_nested_dict({
                'scan': {
                    'on': None,
                    'off': None
                },
                'advertise': {
                    'on': None,
                    'off': None
                },
                'rssi': {
                    'on': None,
                    'off': None
                },
                'show': {
                    'scan': None,
                    'services': None,
                    'attributes': None,
                    'log': None
                },
                'filter': {
                    'address': None,
                },
                'connect': LiveCompleter(self.known_addresses),
                'update-parameters': None,
                'encrypt': None,
                'disconnect': None,
                'discover': {
                    'services': None,
                    'attributes': None
                },
                'request-mtu': None,
                'read': LiveCompleter(self.known_attributes),
                'write': LiveCompleter(self.known_attributes),
                'subscribe': LiveCompleter(self.known_attributes),
                'unsubscribe': LiveCompleter(self.known_attributes),
                'set-phy': {
                    '1m': None,
                    '2m': None,
                    'coded': None
                },
                'set-default-phy': None,
                'quit': None,
                'exit': None
            })

        self.input_field = TextArea(
            height=1,
            prompt="> ",
            multiline=False,
            wrap_lines=False,
            completer=make_completer(),
            history=FileHistory(os.path.join(BUMBLE_USER_DIR, 'history'))
        )

        self.input_field.accept_handler = self.accept_input

        self.output_height = Dimension(min=7, max=7, weight=1)
        self.output_lines = []
        self.output = FormattedTextControl(get_cursor_position=lambda: Point(0, max(0, len(self.output_lines) - 1)))
        self.output_max_lines = 20
        self.scan_results_text = FormattedTextControl()
        self.services_text = FormattedTextControl()
        self.attributes_text = FormattedTextControl()
        self.log_text = FormattedTextControl(get_cursor_position=lambda: Point(0, max(0, len(self.log_lines) - 1)))
        self.log_height = Dimension(min=7, weight=4)
        self.log_max_lines = 100
        self.log_lines = []

        container = HSplit([
            ConditionalContainer(
                Frame(Window(self.scan_results_text), title='Scan Results'),
                filter=Condition(lambda: self.top_tab == 'scan')
            ),
            ConditionalContainer(
                Frame(Window(self.services_text), title='Services'),
                filter=Condition(lambda: self.top_tab == 'services')
            ),
            ConditionalContainer(
                Frame(Window(self.attributes_text), title='Attributes'),
                filter=Condition(lambda: self.top_tab == 'attributes')
            ),
            ConditionalContainer(
                Frame(Window(self.log_text, height=self.log_height), title='Log'),
                filter=Condition(lambda: self.top_tab == 'log')
            ),
            Frame(Window(self.output, height=self.output_height)),
            FormattedTextToolbar(text=self.get_status_bar_text, style='reverse'),
            self.input_field
        ])

        container = FloatContainer(
            container,
            floats=[
                Float(
                    xcursor=True,
                    ycursor=True,
                    content=CompletionsMenu(max_height=16, scroll_offset=1),
                ),
            ],
        )

        layout = Layout(container, focused_element=self.input_field)

        kb = KeyBindings()
        @kb.add("c-c")
        @kb.add("c-q")
        def _(event):
            event.app.exit()

        self.ui = Application(
            layout=layout,
            style=style,
            key_bindings=kb,
            full_screen=True
        )

    async def run_async(self, device_config, transport):
        rssi_monitoring_task = asyncio.create_task(self.rssi_monitor_loop())

        async with await open_transport_or_link(transport) as (hci_source, hci_sink):
            if device_config:
                self.device = Device.from_config_file_with_hci(device_config, hci_source, hci_sink)
            else:
                random_address = f"{random.randint(192,255):02X}"  # address is static random
                for c in random.sample(range(255), 5):
                    random_address += f":{c:02X}"
                self.append_to_log(f"Setting random address: {random_address}")
                self.device = Device.with_hci('Bumble', random_address, hci_source, hci_sink)
            self.device.listener = DeviceListener(self)
            await self.device.power_on()

            # Run the UI
            await self.ui.run_async()

{'fraction_non_alphanumeric': 0.18187686196623634, 'fraction_numerical': 0.00893743793445879, 'mean_word_length': 1.498263458198958, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2919308', 'n_tokens_mistral': 5541, 'n_tokens_neox': 5244, 'n_words': 1120}