| # Copyright 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 |
| # |
| # http://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. |
| |
| """Module for common command handler.""" |
| from typing import Any, Dict |
| |
| from gazoo_device import errors |
| |
| from local_agent import logger as logger_module |
| from local_agent.translation_layer.command_handlers import base |
| |
| |
| logger = logger_module.get_logger() |
| |
| PWRPC_COMMON_CAPABILITY = 'pw_rpc_common' |
| |
| |
| class CommonCommandHandler(base.BaseCommandHandler): |
| """Common command handler. |
| |
| The handler dealing with the device common operations which includes: |
| reboot, factory-reset, OTA. Note that OTA is currently not implemented |
| in the Pigweed RPC endpoint. |
| |
| Smart Home Reboot Trait Schema: |
| https://developers.google.com/assistant/smarthome/traits/reboot |
| """ |
| |
| _REBOOT = 'setReboot' |
| _FACTORY_RESET = 'setFactoryReset' |
| SUPPORTED_METHODS = {_REBOOT, _FACTORY_RESET} |
| |
| def _set_reboot(self, params: Dict[str, Any]) -> None: |
| """Reboots the device. |
| |
| Raises: |
| DeviceError: rebooting fails. |
| """ |
| del params |
| try: |
| self.dut.reboot() |
| except errors.DeviceError as exc: |
| logger.exception(f'Rebooting {self.dut.name} failed.') |
| raise exc |
| |
| def _set_factory_reset(self, params: Dict[str, Any]) -> None: |
| """Factory resets the device. |
| |
| Raises: |
| DeviceError: factory resetting fails. |
| """ |
| del params |
| try: |
| self.dut.factory_reset() |
| except errors.DeviceError as exc: |
| logger.exception(f'Factory resetting {self.dut.name} failed.') |
| raise exc |