How to use is_device_available method in robotframework-ioslibrary

Best Python code snippet using robotframework-ioslibrary_python

framework_adb.py

Source:framework_adb.py

...45        device = None46        log_error("[ADB] installation seems incomplete, adb-shell[usb] is missing (or not working as intended) or adb-server is still running on your system", logger="debuglog")47    if device is not None:48        device.connect(rsa_keys=[signer], auth_timeout_s=30)49    if not is_device_available():50        return51    device_is_connected = True52    log_info(f"[ADB] connected to USB-Device", logger="debuglog")53    update_device_properties()54def disconnect_device() -> NoReturn:55    global device_is_connected, device, device_packages56    device.close()57    device = None58    device_packages = pd.DataFrame(columns=cfg.package_columns + cfg.debloat_columns)59    device_is_connected = False60    log_info(f"[ADB] disconnected from Device", logger="debuglog")61def is_device_available() -> bool:62    global device63    if device is None:64        return False65    is_available = device.available66    # TODO: add additional avail test with: try catch for int(_device.shell("getprop ro.build.version.sdk").strip("\n\r")) > 067    return is_available68def pull_file_from_device(_remote_file_path: str, _local_file_path: str) -> int:69    global device70    if not is_device_available():71        return 072    mode, size, mtime = device.stat(_remote_file_path)73    if size > 0:74        device.pull(_remote_file_path, _local_file_path)75        log_info(f"[ADB] pulled file '{_remote_file_path}' from device, size = {size} byte", logger="debuglog")76    else:77        log_error(f"[ADB] failed pulling file ({_remote_file_path})", logger="debuglog")78    return size79def push_device_package_list_backup(_local_file_path: str, _remote_file_path: str) -> bool:80    global device81    if not os.path.exists(_local_file_path) or not is_device_available():82        return False83    device.push(_local_file_path, _remote_file_path)84    log_info(f"[ADB] pushed file to device ({_remote_file_path})", logger="debuglog")85    mode, size, mtime = device.stat(_remote_file_path)86    return size > 087def restore_device_package_list() -> pd.DataFrame:88    size = pull_file_from_device(cfg.remote_package_file_path, cfg.local_backup_package_file_path)89    if size > 0:90        return misc.load_dataframe(cfg.local_backup_package_file_path)91    else:92        return pd.DataFrame(columns=cfg.package_columns + cfg.debloat_columns[1:3])93def backup_device_package_list() -> bool:94    global device_packages95    packages_filtered = device_packages[cfg.package_columns + cfg.debloat_columns[1:3]]96    misc.save_dataframe(packages_filtered, cfg.local_backup_package_file_path)97    push_device_package_list_backup(cfg.local_backup_package_file_path, cfg.remote_package_file_path)98    return True99def update_device_properties() -> int:100    global device, device_name, device_android_sdk, device_android_version, device_user101    if not is_device_available():102        return 0103    manufacturer = device.shell("getprop ro.product.manufacturer").strip("\n\r")104    product_name = device.shell("getprop ro.product.device").strip("\n\r")105    device_name = manufacturer + " " + product_name106    device_android_version = device.shell("getprop ro.build.version.release").strip("\n\r")107    device_android_sdk = int(device.shell("getprop ro.build.version.sdk").strip("\n\r"))108    log_info(f"[ADB] read device properties, '{device_name}', android {device_android_version}, sdk={device_android_sdk}", logger="debuglog")109    users = get_device_users()110    if len(users) > 1:111        log_info("[ADB] NOTE - there are several users, will choose first one in list!", logger="debuglog")112        log_info(str(users), logger="debuglog")113    device_user = users["index"].iloc[0]114    if device_android_sdk < 26:115        log_error("[ADB] Your android version is old (< 8.0).", logger="debuglog")116        log_error("[ADB] Uninstalled packages can't be restored.", logger="debuglog")117        log_error("[ADB] The GUI won't stop you from doing so.", logger="debuglog")118    return device_android_sdk119def get_device_users() -> pd.DataFrame:120    global device121    user_columns = ["index", "name", "unknown"]122    if not is_device_available():123        return pd.DataFrame(columns=user_columns)124    response = device.shell("pm list users")125    user_list = list([])126    for line in response.splitlines():127        if "UserInfo{" not in line:128            continue129        if "} running" not in line:130            continue131        start = line.find("{") + 1132        end = line.find("}")133        user_list.append(pd.DataFrame([line[start:end].split(":")], columns=user_columns))134    log_info(f"[ADB] read user list ({len(user_list)} entries)", logger="debuglog")135    if len(user_list) > 0:136        user_df = pd.concat(user_list, ignore_index=True).sort_values(by="index", ascending=True).reset_index(drop=True)137    else:138        user_df = pd.DataFrame(columns=user_columns)139    return user_df140def update_package_data() -> NoReturn:141    global device_packages142    package_data1 = device_packages if (len(device_packages) > 0) else restore_device_package_list()143    package_data1["via_adb"] = False144    package_data2 = read_package_data()145    package_data3 = pd.concat([package_data1, package_data2], ignore_index=True)146    package_data4 = package_data3.sort_values(by="via_adb", ascending=False).groupby("package").first().reset_index()147    device_packages = uad_fw.enrich_package_list(package_data4)148    log_info(f"[ADB] read package list ("149             f"{len(device_packages)} entries, "150             f"{len(device_packages[device_packages['via_adb'] == False])} via backup / not ADB)",151             logger="debuglog")152def read_package_data() -> pd.DataFrame:153    global device154    # TODO: vary query by android version155    if not is_device_available():156        return pd.DataFrame(columns=cfg.package_columns + cfg.debloat_columns)157    data = pd.DataFrame(columns=cfg.package_option_names)158    for option in cfg.package_options:159        opt_name = option[1]160        response = device.shell(cfg.package_list_query + option[0])161        for line in response.splitlines():162            if line[0:8].lower() != "package:":163                continue164            package = line[8:]165            data.loc[package, opt_name] = True166    data = data.fillna(False)167    data.index.name = "package"168    data = data.sort_index().reset_index()169    # reformat for more usable meta-data: system package | 3rd-party (user), enabled | disabled, installed | uninstalled170    data.loc[:, "type"] = (data["sys_EN_pUninst"] | data["sys_DIS_pUninst"])  # True == System171    data.loc[:, "enabled"] = data["sys_EN_pUninst"] | data["3rd_EN_pUninst"]172    data.loc[:, "installed"] = data["sys_EN"] | data["sys_DIS"] | data["3rd_EN"] | data["3rd_DIS"]173    data.loc[:, "via_adb"] = True174    return data[cfg.package_columns]175# adb halt, disable, uninstall program176def disable_package(package: str, with_uninstall: bool = False) -> bool:177    global device, device_user, device_android_sdk, device_name, device_packages178    if not is_device_available():179        return False180    # TODO: first find it, do nothing otherwise181    # sdk > 21 seems to need "--user 0"182    cmd_option = f"--user {device_user}" if device_android_sdk > 21 else ""183    # TODO: it could be that "am stopservice" is the new one >= marshmellow/23, the other one <= jelly bean184    cmd1 = f"am force-stop {package}"185    cmd2 = f"pm disable-user {cmd_option} {package}"186    cmd3 = f"pm uninstall -k {cmd_option} {package}"  # "-k" implies to NOT delete user data187    cmd4 = f"pm clear {cmd_option} {package}"188    response1 = device.shell(cmd1)189    response2 = device.shell(cmd2)190    print(f"-> stopped  package '{package}' with response '{response1}'")191    print(f"-> disabled package '{package}' with response '{response2}'")192    misc.save_to_log(device_name, package, cmd2, response2)193    device_packages.loc[device_packages["package"] == package, "enabled"] = False194    time.sleep(cfg.adb_sleep_time_s)195    if with_uninstall:196        response3 = device.shell(cmd3)197        time.sleep(cfg.adb_sleep_time_s)198        response4 = device.shell(cmd4)199        print(f"-> uninstalled package '{package}' with response '{response3}'")200        print(f"-> cleared userdata of package '{package}' with response '{response4}'")201        misc.save_to_log(device_name, package, cmd3, response3)202        misc.save_to_log(device_name, package, cmd4, response4)203        device_packages.loc[device_packages["package"] == package, "installed"] = False204    return True205def enable_package(package: str, with_install: bool = False) -> bool:206    global device, device_name, device_packages207    if not is_device_available():208        return False209    cmd1 = f"cmd package install-existing {package}"210    cmd2 = f"pm enable {package}"211    cmd3 = f"am startservice {package}"212    # TODO: sdk26+ requires services in the foreground "am start-foreground-service"213    if with_install:214        response1 = device.shell(cmd1)215        print(f"-> installed package '{package}' with response '{response1}'")216        misc.save_to_log(device_name, package, cmd1, response1)217        device_packages.loc[device_packages["package"] == package, "installed"] = True218        time.sleep(cfg.adb_sleep_time_s)219    response2 = device.shell(cmd2)220    response3 = device.shell(cmd3)221    print(f"-> enabled package '{package}' with response '{response2}'")  # TODO: parse and process response...

util_distribution.py

Source:util_distribution.py

1# Copyright (c) OpenMMLab. All rights reserved.2import torch3from antgo.framework.helper.parallel import MMDataParallel, MMDistributedDataParallel4dp_factory = {'cuda': MMDataParallel, 'cpu': MMDataParallel}5ddp_factory = {'cuda': MMDistributedDataParallel}6def build_dp(model, device='cuda', dim=0, *args, **kwargs):7    """build DataParallel module by device type.8    if device is cuda, return a MMDataParallel model; if device is mlu,9    return a MLUDataParallel model.10    Args:11        model (:class:`nn.Module`): model to be parallelized.12        device (str): device type, cuda, cpu or mlu. Defaults to cuda.13        dim (int): Dimension used to scatter the data. Defaults to 0.14    Returns:15        nn.Module: the model to be parallelized.16    """17    if device == 'cuda':18        model = model.cuda()19    return dp_factory[device](model, dim=dim, *args, **kwargs)20def build_ddp(model, device='cuda', *args, **kwargs):21    """Build DistributedDataParallel module by device type.22    If device is cuda, return a MMDistributedDataParallel model;23    if device is mlu, return a MLUDistributedDataParallel model.24    Args:25        model (:class:`nn.Module`): module to be parallelized.26        device (str): device type, mlu or cuda.27    Returns:28        :class:`nn.Module`: the module to be parallelized29    References:30        .. [1] https://pytorch.org/docs/stable/generated/torch.nn.parallel.31                     DistributedDataParallel.html32    """33    assert device in ['cuda', 'mlu'], 'Only available for cuda or mlu devices.'34    if device == 'cuda':35        model = model.cuda()36    return ddp_factory[device](model, *args, **kwargs)37def is_mlu_available():38    """Returns a bool indicating if MLU is currently available."""39    return hasattr(torch, 'is_mlu_available') and torch.is_mlu_available()40def get_device():41    """Returns an available device, cpu, cuda or mlu."""42    is_device_available = {43        'cuda': torch.cuda.is_available(),44        'mlu': is_mlu_available()45    }46    device_list = [k for k, v in is_device_available.items() if v]...

client.py

Source:client.py

...16        return raw_device_list17    def available_devices(self) -> typing.List[typing.List[str]]:18        device_list = self.devices()19        return [each for each in device_list if each[1] == "device"]20    def is_device_available(self, serial_no: str) -> bool:21        return serial_no in [each[0] for each in self.available_devices()]22    def kill_server(self):23        return run_cmd(["adb", "kill-server"])24    def start_server(self):25        return run_cmd(["adb", "start-server"])26    def restart_server(self):27        self.kill_server()28        self.start_server()29if __name__ == "__main__":30    cli = ADBClient()31    cli.restart_server()32    d = cli.available_devices()33    print(d)34    a = cli.is_device_available("123456F")...

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.