"""
WebSocket JSON-based server that exposes _ToyHub to connected clients.
Offers an alternative to the Low-Level / High-Level API defined by the tikal library.
Here information is exchanged via a websocket. Significantly harder to use than the Low-Level / High-Level APIs but
offers some advantages:
- Process separation
- Service can be used in applications written in other programming languages (assuming websockets are supported)
- Multiple clients can modify the same state (experimental, untested)
This module is the entry point to the ToyServer command-line-interface.
"""
# nuitka-project: --msvc=latest
# nuitka-project: --mode=standalone
# nuitka-project: --include-windows-runtime-dlls=yes
# nuitka-project: --windows-console-mode=disable
# nuitka-project: --include-package=bleak
# nuitka-project: --include-package=bleak.backends.winrt
# nuitka-project: --include-package=winrt._winrt
# nuitka-project: --include-package=winrt._winrt_windows_devices_bluetooth
# nuitka-project: --include-package=winrt._winrt_windows_devices_bluetooth_advertisement
# nuitka-project: --include-package=winrt._winrt_windows_devices_bluetooth_genericattributeprofile
# nuitka-project: --include-package=winrt._winrt_windows_devices_enumeration
# nuitka-project: --include-package=winrt._winrt_windows_devices_radios
# nuitka-project: --include-package=winrt._winrt_windows_foundation
# nuitka-project: --include-package=winrt._winrt_windows_foundation_collections
# nuitka-project: --include-package=winrt._winrt_windows_storage_streams
# nuitka-project: --include-package=winrt.runtime
# nuitka-project: --include-package=winrt.runtime._internals
# nuitka-project: --include-package=winrt.runtime.interop
# nuitka-project: --include-package=winrt.system
# nuitka-project: --include-package=winrt.system.hresult
import argparse
import asyncio
import logging
import traceback
from pathlib import Path
from .toy_server import ToyServer
[docs]
def main() -> None:
"""
Entry point for the ToyServer command-line interface.
Parses command-line arguments, configures logging, constructs a ToyServer instance.
ToyServer shuts down automatically if no client is connected for 3 seconds.
Command-line arguments:
--host: Host to bind to (default: localhost).
--port: Port to listen on (default: 8142).
--toy-cache-path: Path to the toy-cache file (default: ./data/toy_cache.json).
--mock-toys: Use a software mock instead of real Bluetooth hardware.
--log-path: Filepath to write logs to (default: ./data/tikal_ws.log).
--log-level: Logging verbosity: DEBUG, INFO, WARNING, or ERROR (default: INFO).
"""
parser = argparse.ArgumentParser(description="WebSocket server for _ToyHub")
parser.add_argument(
"--host", default="localhost", help="Host to bind to (default: localhost)"
)
parser.add_argument(
"--port", type=int, default=8142, help="Port to listen on (default: 8142)"
)
parser.add_argument(
"--timeout",
type=int,
default=3,
help="If no client is connected for this many seconds, the server will shut down automatically (default: 3 seconds). Set to 0 to disable auto-shutdown.",
)
parser.add_argument(
"--toy-cache-path",
type=Path,
default=Path("./data/toy_cache.json"),
help="Path to toy cache file (default: ./data/toy_cache.json). If the string 'None' is passed uses in-memory cache only.",
)
parser.add_argument(
"--mock-toys",
action="store_true",
help="Use mock toys instead of real Bluetooth",
)
parser.add_argument(
"--log-path",
default="./data/tikal_ws.log",
help="File to write the log to (default: ./data/tikal_ws.log). If the string 'None' is passed disables logging.",
)
parser.add_argument(
"--log-level",
default="INFO",
help="Logging level (DEBUG, INFO, WARNING, ERROR)",
)
args = parser.parse_args()
formatting = logging.Formatter(
"%(asctime)s [%(levelname)s] : %(module)s.%(funcName)s reports: %(message)s"
)
logger = logging.getLogger("tikal_ws")
if args.log_path != "None":
log_path = Path(args.log_path)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(Path(args.log_path), "w", "utf-8")
file_handler.setLevel(args.log_level)
file_handler.setFormatter(formatting)
logger.setLevel(args.log_level.upper())
logger.addHandler(file_handler)
logger.info("Starting ToyServer")
toy_cache_path = (
Path(args.toy_cache_path) if args.toy_cache_path != "None" else Path()
)
server = ToyServer(
toy_cache_path=toy_cache_path,
host=args.host,
port=args.port,
idle_shutdown_delay=args.timeout,
mock_toys=args.mock_toys,
)
try:
asyncio.run(server.serve())
except Exception:
details = traceback.format_exc()
logging.critical("Server shutting down due to unhandled exception: %s", details)
if __name__ == "__main__":
main()