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 logging
18import json
19
20from bumble import pandora as bumble_server
21from bumble.pandora import PandoraDevice, Config, serve
22
23from bumble_experimental.asha import AshaService
24from bumble_experimental.dck import DckService
25from bumble_experimental.gatt import GATTService
26from bumble_experimental.rfcomm import RFCOMMService
27
28from pandora_experimental.asha_grpc_aio import add_AshaServicer_to_server
29from pandora_experimental.dck_grpc_aio import add_DckServicer_to_server
30from pandora_experimental.gatt_grpc_aio import add_GATTServicer_to_server
31from pandora_experimental.rfcomm_grpc_aio import add_RFCOMMServicer_to_server
32
33from typing import Dict, Any
34
35BUMBLE_SERVER_GRPC_PORT = 7999
36ROOTCANAL_PORT_CUTTLEFISH = 7300
37
38
39def main(grpc_port: int, rootcanal_port: int, transport: str, config: str) -> None:
40    register_experimental_services()
41    if '<rootcanal-port>' in transport:
42        transport = transport.replace('<rootcanal-port>', str(rootcanal_port))
43
44    bumble_config = retrieve_config(config)
45    bumble_config.setdefault('transport', transport)
46    device = PandoraDevice(bumble_config)
47
48    server_config = Config()
49    server_config.load_from_dict(bumble_config.get('server', {}))
50
51    logging.basicConfig(level=logging.DEBUG,
52                        format='%(asctime)s.%(msecs).03d %(levelname)-8s %(message)s',
53                        datefmt='%m-%d %H:%M:%S')
54    asyncio.run(serve(device, config=server_config, port=grpc_port))
55
56
57def args_parser() -> argparse.ArgumentParser:
58    parser = argparse.ArgumentParser(description="Bumble command-line tool")
59
60    parser.add_argument('--grpc-port', type=int, default=BUMBLE_SERVER_GRPC_PORT, help='gRPC port to serve')
61    parser.add_argument('--rootcanal-port', type=int, default=ROOTCANAL_PORT_CUTTLEFISH, help='Rootcanal TCP port')
62    parser.add_argument('--transport',
63                        type=str,
64                        default='tcp-client:127.0.0.1:<rootcanal-port>',
65                        help='HCI transport (default: tcp-client:127.0.0.1:<rootcanal-port>)')
66    parser.add_argument('--config', type=str, help='Bumble json configuration file')
67
68    return parser
69
70
71def register_experimental_services():
72    bumble_server.register_servicer_hook(
73        lambda bumble, _, server: add_AshaServicer_to_server(AshaService(bumble.device), server))
74    bumble_server.register_servicer_hook(
75        lambda bumble, _, server: add_DckServicer_to_server(DckService(bumble.device), server))
76    bumble_server.register_servicer_hook(
77        lambda bumble, _, server: add_GATTServicer_to_server(GATTService(bumble.device), server))
78    bumble_server.register_servicer_hook(
79        lambda bumble, _, server: add_RFCOMMServicer_to_server(RFCOMMService(bumble.device), server))
80
81
82def retrieve_config(config: str) -> Dict[str, Any]:
83    if not config:
84        return {}
85
86    with open(config, 'r') as f:
87        return json.load(f)
88
89
90if __name__ == '__main__':
91    args = args_parser().parse_args()
92    main(**vars(args))
93