How to use _running_apps method in pyatom

Best Python code snippet using pyatom_python

firetv.py

Source:firetv.py

1"""2Support for functionality to interact with FireTV devices.3For more details about this platform, please refer to the documentation at4https://home-assistant.io/components/media_player.firetv/5"""6import functools7import logging8import threading9import voluptuous as vol10from homeassistant.components.media_player import (11    MediaPlayerDevice, PLATFORM_SCHEMA, SUPPORT_NEXT_TRACK, SUPPORT_PAUSE,12    SUPPORT_PLAY, SUPPORT_PREVIOUS_TRACK, SUPPORT_SELECT_SOURCE, SUPPORT_STOP,13    SUPPORT_TURN_OFF, SUPPORT_TURN_ON, SUPPORT_VOLUME_SET, )14from homeassistant.const import (15    CONF_HOST, CONF_NAME, CONF_PORT, STATE_IDLE, STATE_OFF, STATE_PAUSED,16    STATE_PLAYING, STATE_STANDBY)17import homeassistant.helpers.config_validation as cv18REQUIREMENTS = ['firetv==1.0.7']19_LOGGER = logging.getLogger(__name__)20SUPPORT_FIRETV = SUPPORT_PAUSE | \21    SUPPORT_TURN_ON | SUPPORT_TURN_OFF | SUPPORT_PREVIOUS_TRACK | \22    SUPPORT_NEXT_TRACK | SUPPORT_SELECT_SOURCE | SUPPORT_STOP | \23    SUPPORT_VOLUME_SET | SUPPORT_PLAY24CONF_ADBKEY = 'adbkey'25CONF_GET_SOURCE = 'get_source'26CONF_GET_SOURCES = 'get_sources'27DEFAULT_NAME = 'Amazon Fire TV'28DEFAULT_PORT = 555529DEFAULT_GET_SOURCE = True30DEFAULT_GET_SOURCES = True31def has_adb_files(value):32    """Check that ADB key files exist."""33    priv_key = value34    pub_key = '{}.pub'.format(value)35    cv.isfile(pub_key)36    return cv.isfile(priv_key)37PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({38    vol.Required(CONF_HOST): cv.string,39    vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,40    vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,41    vol.Optional(CONF_ADBKEY): has_adb_files,42    vol.Optional(CONF_GET_SOURCE, default=DEFAULT_GET_SOURCE): cv.boolean,43    vol.Optional(CONF_GET_SOURCES, default=DEFAULT_GET_SOURCES): cv.boolean44})45PACKAGE_LAUNCHER = "com.amazon.tv.launcher"46PACKAGE_SETTINGS = "com.amazon.tv.settings"47def setup_platform(hass, config, add_entities, discovery_info=None):48    """Set up the FireTV platform."""49    from firetv import FireTV50    host = '{0}:{1}'.format(config[CONF_HOST], config[CONF_PORT])51    if CONF_ADBKEY in config:52        ftv = FireTV(host, config[CONF_ADBKEY])53        adb_log = " using adbkey='{0}'".format(config[CONF_ADBKEY])54    else:55        ftv = FireTV(host)56        adb_log = ""57    if not ftv.available:58        _LOGGER.warning("Could not connect to Fire TV at %s%s", host, adb_log)59        return60    name = config[CONF_NAME]61    get_source = config[CONF_GET_SOURCE]62    get_sources = config[CONF_GET_SOURCES]63    device = FireTVDevice(ftv, name, get_source, get_sources)64    add_entities([device])65    _LOGGER.info("Setup Fire TV at %s%s", host, adb_log)66def adb_decorator(override_available=False):67    """Send an ADB command if the device is available and not locked."""68    def adb_wrapper(func):69        """Wait if previous ADB commands haven't finished."""70        @functools.wraps(func)71        def _adb_wrapper(self, *args, **kwargs):72            # If the device is unavailable, don't do anything73            if not self.available and not override_available:74                return None75            # If an ADB command is already running, skip this command76            if not self.adb_lock.acquire(blocking=False):77                _LOGGER.info('Skipping an ADB command because a previous '78                             'command is still running')79                return None80            # Additional ADB commands will be prevented while trying this one81            try:82                returns = func(self, *args, **kwargs)83            except self.exceptions:84                _LOGGER.error('Failed to execute an ADB command; will attempt '85                              'to re-establish the ADB connection in the next '86                              'update')87                returns = None88                self._available = False  # pylint: disable=protected-access89            finally:90                self.adb_lock.release()91            return returns92        return _adb_wrapper93    return adb_wrapper94class FireTVDevice(MediaPlayerDevice):95    """Representation of an Amazon Fire TV device on the network."""96    def __init__(self, ftv, name, get_source, get_sources):97        """Initialize the FireTV device."""98        from adb.adb_protocol import (99            InvalidChecksumError, InvalidCommandError, InvalidResponseError)100        self.firetv = ftv101        self._name = name102        self._get_source = get_source103        self._get_sources = get_sources104        # whether or not the ADB connection is currently in use105        self.adb_lock = threading.Lock()106        # ADB exceptions to catch107        self.exceptions = (AttributeError, BrokenPipeError, TypeError,108                           ValueError, InvalidChecksumError,109                           InvalidCommandError, InvalidResponseError)110        self._state = None111        self._available = self.firetv.available112        self._current_app = None113        self._running_apps = None114    @property115    def name(self):116        """Return the device name."""117        return self._name118    @property119    def should_poll(self):120        """Device should be polled."""121        return True122    @property123    def supported_features(self):124        """Flag media player features that are supported."""125        return SUPPORT_FIRETV126    @property127    def state(self):128        """Return the state of the player."""129        return self._state130    @property131    def available(self):132        """Return whether or not the ADB connection is valid."""133        return self._available134    @property135    def source(self):136        """Return the current app."""137        return self._current_app138    @property139    def source_list(self):140        """Return a list of running apps."""141        return self._running_apps142    @adb_decorator(override_available=True)143    def update(self):144        """Get the latest date and update device state."""145        # Check if device is disconnected.146        if not self._available:147            self._running_apps = None148            self._current_app = None149            # Try to connect150            self.firetv.connect()151            self._available = self.firetv.available152        # If the ADB connection is not intact, don't update.153        if not self._available:154            return155        # Check if device is off.156        if not self.firetv.screen_on:157            self._state = STATE_OFF158            self._running_apps = None159            self._current_app = None160        # Check if screen saver is on.161        elif not self.firetv.awake:162            self._state = STATE_IDLE163            self._running_apps = None164            self._current_app = None165        else:166            # Get the running apps.167            if self._get_sources:168                self._running_apps = self.firetv.running_apps169            # Get the current app.170            if self._get_source:171                current_app = self.firetv.current_app172                if isinstance(current_app, dict)\173                        and 'package' in current_app:174                    self._current_app = current_app['package']175                else:176                    self._current_app = current_app177                # Show the current app as the only running app.178                if not self._get_sources:179                    if self._current_app:180                        self._running_apps = [self._current_app]181                    else:182                        self._running_apps = None183                # Check if the launcher is active.184                if self._current_app in [PACKAGE_LAUNCHER,185                                         PACKAGE_SETTINGS]:186                    self._state = STATE_STANDBY187                # Check for a wake lock (device is playing).188                elif self.firetv.wake_lock:189                    self._state = STATE_PLAYING190                # Otherwise, device is paused.191                else:192                    self._state = STATE_PAUSED193            # Don't get the current app.194            elif self.firetv.wake_lock:195                # Check for a wake lock (device is playing).196                self._state = STATE_PLAYING197            else:198                # Assume the devices is on standby.199                self._state = STATE_STANDBY200    @adb_decorator()201    def turn_on(self):202        """Turn on the device."""203        self.firetv.turn_on()204    @adb_decorator()205    def turn_off(self):206        """Turn off the device."""207        self.firetv.turn_off()208    @adb_decorator()209    def media_play(self):210        """Send play command."""211        self.firetv.media_play()212    @adb_decorator()213    def media_pause(self):214        """Send pause command."""215        self.firetv.media_pause()216    @adb_decorator()217    def media_play_pause(self):218        """Send play/pause command."""219        self.firetv.media_play_pause()220    @adb_decorator()221    def media_stop(self):222        """Send stop (back) command."""223        self.firetv.back()224    @adb_decorator()225    def volume_up(self):226        """Send volume up command."""227        self.firetv.volume_up()228    @adb_decorator()229    def volume_down(self):230        """Send volume down command."""231        self.firetv.volume_down()232    @adb_decorator()233    def media_previous_track(self):234        """Send previous track command (results in rewind)."""235        self.firetv.media_previous()236    @adb_decorator()237    def media_next_track(self):238        """Send next track command (results in fast-forward)."""239        self.firetv.media_next()240    @adb_decorator()241    def select_source(self, source):242        """Select input source.243        If the source starts with a '!', then it will close the app instead of244        opening it.245        """246        if isinstance(source, str):247            if not source.startswith('!'):248                self.firetv.launch_app(source)249            else:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.