platypush/plugins/switchbot/__init__.py
def status(self, device: Optional[str] = None) -> Union[dict, List[dict]]:
        """
        Get the status of all the registered devices or of a specific device.

        :param device: Filter by device ID or name.
        :return: .. schema:: switchbot.DeviceStatusSchema(many=True)
        """
        # noinspection PyUnresolvedReferences
        devices = self.devices().output
        if device:
            return {
                **device,
                **self._run('get', 'status', device=self._get_device(device)),
            }

        devices_by_id = {dev['id']: dev for dev in devices}
        queues = [queue.Queue()] * len(devices)
        workers = [
            threading.Thread(target=self._worker, args=(queues[i], 'get', 'status'), kwargs={'device': dev})
            for i, dev in enumerate(devices)
        ]

        results = []
        for worker in workers:
            worker.start()

        for q in queues:
            response = q.get()
            if not response:
                continue

            assert not isinstance(response, Exception), str(response)
            results.append({
                **devices_by_id.get(response.get('id'), {}),
                **response,
            })

        for worker in workers:
            worker.join()

        return results

        Get the status of all the registered devices or of a specific device.

        :param device: Filter by device ID or name.
        :return: .. schema:: switchbot.DeviceStatusSchema(many=True)
        
Get the status of all the registered devices or of a specific device.
Get the status of all the registered devices or of a specific device .
def status(self, device: Optional[str] = None) -> Union[dict, List[dict]]:
        devices = self.devices().output
        if device:
            return {
                **device,
                **self._run('get', 'status', device=self._get_device(device)),
            }
        devices_by_id = {dev['id']: dev for dev in devices}
        queues = [queue.Queue()] * len(devices)
        workers = [
            threading.Thread(target=self._worker, args=(queues[i], 'get', 'status'), kwargs={'device': dev})
            for i, dev in enumerate(devices)
        ]
        results = []
        for worker in workers:
            worker.start()
        for q in queues:
            response = q.get()
            if not response:
                continue
            assert not isinstance(response, Exception), str(response)
            results.append({
                **devices_by_id.get(response.get('id'), {}),
                **response,
            })
        for worker in workers:
            worker.join()
        return results
def status ( self , device : Optional [ str ] = None ) -> Union [ dict , List [ dict ] ] : devices = self . devices ( ) . output if device : return { ** device , ** self . _run ( 'get' , 'status' , device = self . _get_device ( device ) ) , } devices_by_id = { dev [ 'id' ] : dev for dev in devices } queues = [ queue . Queue ( ) ] * len ( devices ) workers = [ threading . Thread ( target = self . _worker , args = ( queues [ i ] , 'get' , 'status' ) , kwargs = { 'device' : dev } ) for i , dev in enumerate ( devices ) ] results = [ ] for worker in workers : worker . start ( ) for q in queues : response = q . get ( ) if not response : continue assert not isinstance ( response , Exception ) , str ( response ) results . append ( { ** devices_by_id . get ( response . get ( 'id' ) , { } ) , ** response , } ) for worker in workers : worker . join ( ) return results
Get the status of all the registered devices or of a specific device.
Get the status of all the registered devices or of a specific device .
"""
        Get the status of all the registered devices or of a specific device.

        :param device: Filter by device ID or name.
        :return: .. schema:: switchbot.DeviceStatusSchema(many=True)
        """ # noinspection PyUnresolvedReferences
None
self
device
Optional[str]
Filter by device ID or name.
Filter by device ID or name .