1# Copyright 2023 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import argparse
16import asyncio
17import json
18import logging
19from typing import Any, Dict
20
21from bumble import pandora as bumble_server
22from bumble.pandora import Config, PandoraDevice, serve
23from bumble_experimental.asha import AshaService
24from bumble_experimental.avrcp import AvrcpService
25from bumble_experimental.bumble_config import BumbleConfigService
26from bumble_experimental.dck import DckService
27from bumble_experimental.gatt import GATTService
28from bumble_experimental.hid import HIDService
29from bumble_experimental.oob import OOBService
30from bumble_experimental.rfcomm import RFCOMMService
31from pandora_experimental.asha_grpc_aio import add_AshaServicer_to_server
32from pandora_experimental.avrcp_grpc_aio import add_AVRCPServicer_to_server
33from pandora_experimental.bumble_config_grpc_aio import \
34    add_BumbleConfigServicer_to_server
35from pandora_experimental.dck_grpc_aio import add_DckServicer_to_server
36from pandora_experimental.gatt_grpc_aio import add_GATTServicer_to_server
37from pandora_experimental.hid_grpc_aio import add_HIDServicer_to_server
38from pandora_experimental.oob_grpc_aio import add_OOBServicer_to_server
39from pandora_experimental.rfcomm_grpc_aio import add_RFCOMMServicer_to_server
40
41BUMBLE_SERVER_GRPC_PORT = 7999
42ROOTCANAL_PORT_CUTTLEFISH = 7300
43
44
45def main(grpc_port: int, rootcanal_port: int, transport: str, config: str) -> None:
46    register_experimental_services()
47    if '<rootcanal-port>' in transport:
48        transport = transport.replace('<rootcanal-port>', str(rootcanal_port))
49
50    bumble_config = retrieve_config(config)
51    bumble_config.setdefault('transport', transport)
52    device = PandoraDevice(bumble_config)
53
54    server_config = Config()
55    server_config.load_from_dict(bumble_config.get('server', {}))
56
57    logging.basicConfig(level=logging.DEBUG,
58                        format='%(asctime)s.%(msecs).03d %(levelname)-8s %(message)s',
59                        datefmt='%m-%d %H:%M:%S')
60    asyncio.run(serve(device, config=server_config, port=grpc_port))
61
62
63def args_parser() -> argparse.ArgumentParser:
64    parser = argparse.ArgumentParser(description="Bumble command-line tool")
65
66    parser.add_argument('--grpc-port', type=int, default=BUMBLE_SERVER_GRPC_PORT, help='gRPC port to serve')
67    parser.add_argument('--rootcanal-port', type=int, default=ROOTCANAL_PORT_CUTTLEFISH, help='Rootcanal TCP port')
68    parser.add_argument('--transport',
69                        type=str,
70                        default='tcp-client:127.0.0.1:<rootcanal-port>',
71                        help='HCI transport (default: tcp-client:127.0.0.1:<rootcanal-port>)')
72    parser.add_argument('--config', type=str, help='Bumble json configuration file')
73
74    return parser
75
76
77def register_experimental_services() -> None:
78    bumble_server.register_servicer_hook(
79        lambda bumble, _, server: add_AVRCPServicer_to_server(AvrcpService(bumble.device), server))
80    bumble_server.register_servicer_hook(
81        lambda bumble, _, server: add_AshaServicer_to_server(AshaService(bumble.device), server))
82    bumble_server.register_servicer_hook(
83        lambda bumble, _, server: add_DckServicer_to_server(DckService(bumble.device), server))
84    bumble_server.register_servicer_hook(
85        lambda bumble, _, server: add_GATTServicer_to_server(GATTService(bumble.device), server))
86    bumble_server.register_servicer_hook(
87        lambda bumble, _, server: add_RFCOMMServicer_to_server(RFCOMMService(bumble.device), server))
88    bumble_server.register_servicer_hook(
89        lambda bumble, _, server: add_HIDServicer_to_server(HIDService(bumble.device), server))
90    bumble_server.register_servicer_hook(
91        lambda bumble, _, server: add_OOBServicer_to_server(OOBService(bumble.device), server))
92    bumble_server.register_servicer_hook(
93        lambda bumble, config, server: add_BumbleConfigServicer_to_server(BumbleConfigService(bumble.device, config), server))
94
95
96def retrieve_config(config: str) -> Dict[str, Any]:
97    if not config:
98        return {}
99
100    with open(config, 'r') as f:
101        return json.load(f)
102
103
104if __name__ == '__main__':
105    args = args_parser().parse_args()
106    main(**vars(args))
107