Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_machine_info method in pytest-benchmark

Best Python code snippet using pytest-benchmark

device_server.py

Source:device_server.py

...118        119    data = bottle.request.json120    update_machine_json_data.update(data['machine_options']['arguments'])121    122    if 'node_ip' in update_machine_json_data and update_machine_json_data['node_ip'] != get_machine_info(id)['etc_node_ip']:123        set_etc_hostname(update_machine_json_data['node_ip'])124        haschanged = True125    126    if 'etho_number' in update_machine_json_data and int(update_machine_json_data['etho_number']) != int(get_machine_info(id)['machine-number']):127        set_machine_name(update_machine_json_data['etho_number'])128        set_machine_id(update_machine_json_data['etho_number'])129        haschanged = True130    131    if 'ESSID' in update_machine_json_data and 'Key' in update_machine_json_data and (update_machine_json_data['ESSID'] != get_machine_info(id)['WIFI_SSID'] or update_machine_json_data['Key'] != get_machine_info(id)['WIFI_PASSWORD']):132        set_WIFI(ssid=update_machine_json_data['ESSID'], wpakey=update_machine_json_data['Key'])133        haschanged = True134    if 'isexperimental' in update_machine_json_data and update_machine_json_data['isexperimental'] != isExperimental():135        isExperimental(update_machine_json_data['isexperimental'])136        haschanged = True137    138    #Time comes as number of milliseconds from timestamp139    if 'datetime' in update_machine_json_data and update_machine_json_data['datetime']:140        tn = datetime.datetime.fromtimestamp(update_machine_json_data['datetime'])141        set_datetime(tn)142        143    return {"haschanged": haschanged}144    #return get_machine_info(id)145    146@api.post('/controls/<id>/<action>')147@error_decorator148def controls(id, action):149    global control150    global record151    if id != machine_id:152        raise WrongMachineID153    if action == 'start':154        data = bottle.request.json155        tracking_json_data.update(data)156        157        control = None158        control = ControlThread(machine_id=machine_id,159                                name=machine_name,160                                version=version,161                                ethoscope_dir=ETHOSCOPE_DIR,162                                data=tracking_json_data)163        control.start()164        return info(id)165    elif action in ['stop', 'close', 'poweroff', 'reboot', 'restart']:166        167        if control.info['status'] in ['running', 'recording', 'streaming'] :168            logging.info("Stopping monitor")169            control.stop()170            logging.info("Joining monitor")171            control.join()172            logging.info("Monitor joined")173            logging.info("Monitor stopped")174        if action == 'close':175            close()176        if action == 'poweroff':177            logging.info("Stopping monitor due to poweroff request")178            logging.info("Powering off Device.")179            subprocess.call('poweroff')180        if action == 'reboot':181            logging.info("Stopping monitor due to reboot request")182            logging.info("Powering off Device.")183            subprocess.call('reboot')184        if action == 'restart':185            logging.info("Restarting service")186            subprocess.call(['systemctl', 'restart', 'ethoscope_device'])187        return info(id)188    elif action in ['start_record', 'stream']:189        data = bottle.request.json190        recording_json_data.update(data)191        logging.warning("Recording or Streaming video, data is %s" % str(data))192        control = None193        control = ControlThreadVideoRecording(machine_id=machine_id,194                                              name=machine_name,195                                              version=version,196                                              ethoscope_dir=ETHOSCOPE_DIR,197                                              data=recording_json_data)198        control.start()199        return info(id)200        201    else:202        raise Exception("No such action: %s" % action)203@api.get('/data/listfiles/<category>/<id>')204@error_decorator205def list_data_files(category, id):206    '''207    provides a list of files in the ethoscope data folders, that were either uploaded or generated208    category is the name of the folder209    this is not meant to report db files or h264 files but it's supposed to be working for things like masks and other user generated files210    '''211    if id != machine_id:212        raise WrongMachineID213    path = os.path.join (ETHOSCOPE_UPLOAD, category)214    if os.path.exists(path):215        return {'filelist' : [{'filename': i, 'fullpath' : os.path.abspath(os.path.join(path,i))} for i in os.listdir(path)]}216    return {}217@api.get('/machine/<id>')218@error_decorator219def get_machine_info(id):220    """221    This is information about the ethoscope that is not changing in time such as hardware specs and configuration parameters222    """223    if id is not None and id != machine_id:224        raise WrongMachineID225    machine_info = {}226    machine_info['node_ip'] = bottle.request.environ.get('HTTP_X_FORWARDED_FOR') or bottle.request.environ.get('REMOTE_ADDR')227    228    try:229        machine_info['etc_node_ip'] = get_etc_hostnames()[NODE]230    except:231        machine_info['etc_node_ip'] = "not set"232    machine_info['knows_node_ip'] = ( machine_info['node_ip'] == machine_info['etc_node_ip'] )233    machine_info['hostname'] = os.uname()[1]234    235    machine_info['machine-name'] = get_machine_name()236    237    try:238        machine_info['machine-number'] = int ( machine_info['machine-name'].split("_")[1] )239    except:240        machine_info['machine-number'] = 0241        242        243    machine_info['machine-id'] = get_machine_id()244    machine_info['kernel'] = os.uname()[2]245    machine_info['pi_version'] = pi_version()246    machine_info['camera'] = getPiCameraVersion()247    248    try:249        machine_info['WIFI_SSID'] = get_WIFI()['ESSID']250    except: 251        machine_info['WIFI_SSID'] = "not set"252    try:    253        machine_info['WIFI_PASSWORD'] = get_WIFI()['Key']254    except:255        machine_info['WIFI_PASSWORD'] = "not set"256    257    machine_info['SD_CARD_AGE'] = get_SD_CARD_AGE()258    machine_info['partitions'] = get_partition_infos()259    260    return machine_info261@api.get('/data/<id>')262@error_decorator263def info(id):264    """265    This is information that is changing in time as the machine operates, such as FPS during tracking, CPU temperature etc266    """267    268    info = {}269    if machine_id != id:270        raise WrongMachineID271    272    if control is not None: 273        info = control.info274        275    info["current_timestamp"] = bottle.time.time()276    info["CPU_temp"] = get_core_temperature()277    return info278@api.get('/user_options/<id>')279@error_decorator280def user_options(id):281    '''282    Passing back options regarding what information can be changed on the the device. This populates the form on the node GUI283    '''284    if machine_id != id:285        raise WrongMachineID286    287    288        289    return {290        "tracking":ControlThread.user_options(),291        "recording":ControlThreadVideoRecording.user_options(),292        "streaming": {},293        "update_machine": { "machine_options": [{"overview": "Machine information that can be set by the user",294                            "arguments": [295                                {"type": "number", "name":"etho_number", "description": "An ID number (1-999) unique to this ethoscope","default": get_machine_info(id)['machine-number'] },296                                {"type": "boolean", "name":"isexperimental", "description": "Specify if the ethoscope is to be treated as experimental", "default": isExperimental()}, 297                                {"type": "str", "name":"node_ip", "description": "The IP address that you want to record as the node (do not change this value unless you know what you are doing!)","default": get_machine_info(id)['node_ip']},298                                {"type": "str", "name":"ESSID", "description": "The name of the WIFI SSID","default": get_machine_info(id)['WIFI_SSID'] },299                                {"type": "str", "name":"Key", "description": "The WPA password for the WIFI SSID","default": get_machine_info(id)['WIFI_PASSWORD'] }],300                            "name" : "Ethoscope Options"}],301                               } }302@api.get('/data/log/<id>')303@error_decorator304def get_log(id):305    '''306    returns the journalctl log307    '''308    output = "No log available"309    try:310        with os.popen('journalctl -u ethoscope_device.service -rb') as p:311            output = p.read()312    except Exception as e:313        logging.error(traceback.format_exc())...

machine_info.py

Source:machine_info.py

...35        else:36            self.logger = logger37        self.machine_info = None38        try:39            self.machine_info = self.get_machine_info()40        except Exception:41            self.logger.exception("Exception in getting machine info.")42            self.machine_info = None43    def get_machine_info(self):44        """Get machine info in metric format"""45        gpu_info = self.get_gpu_info_by_nvml()46        cpu_info = cpuinfo.get_cpu_info()47        machine_info = {48            "gpu": gpu_info,49            "cpu": self.get_cpu_info(),50            "memory": self.get_memory_info(),51            "os": platform.platform(),52            "python": self._try_get(cpu_info, ["python_version"]),53            "packages": self.get_related_packages(),54            "onnxruntime": self.get_onnxruntime_info(),55            "pytorch": self.get_pytorch_info(),56            "tensorflow": self.get_tensorflow_info(),57        }58        return machine_info59    def get_memory_info(self) -> Dict:60        """Get memory info"""61        mem = psutil.virtual_memory()62        return {"total": mem.total, "available": mem.available}63    def _try_get(self, cpu_info: Dict, names: List) -> str:64        for name in names:65            if name in cpu_info:66                value = cpu_info[name]67                if isinstance(value, (list, tuple)):68                    return ",".join([str(i) for i in value])69                return value70        return ""71    def get_cpu_info(self) -> Dict:72        """Get CPU info"""73        cpu_info = cpuinfo.get_cpu_info()74        return {75            "brand": self._try_get(cpu_info, ["brand", "brand_raw"]),76            "cores": psutil.cpu_count(logical=False),77            "logical_cores": psutil.cpu_count(logical=True),78            "hz": self._try_get(cpu_info, ["hz_actual"]),79            "l2_cache": self._try_get(cpu_info, ["l2_cache_size"]),80            "flags": self._try_get(cpu_info, ["flags"]),81            "processor": platform.uname().processor,82        }83    def get_gpu_info_by_nvml(self) -> Dict:84        """Get GPU info using nvml"""85        gpu_info_list = []86        driver_version = None87        try:88            nvmlInit()89            driver_version = nvmlSystemGetDriverVersion()90            deviceCount = nvmlDeviceGetCount()91            for i in range(deviceCount):92                handle = nvmlDeviceGetHandleByIndex(i)93                info = nvmlDeviceGetMemoryInfo(handle)94                gpu_info = {}95                gpu_info["memory_total"] = info.total96                gpu_info["memory_available"] = info.free97                gpu_info["name"] = nvmlDeviceGetName(handle)98                gpu_info_list.append(gpu_info)99            nvmlShutdown()100        except NVMLError as error:101            if not self.silent:102                self.logger.error("Error fetching GPU information using nvml: %s", error)103            return None104        result = {"driver_version": driver_version, "devices": gpu_info_list}105        if "CUDA_VISIBLE_DEVICES" in environ:106            result["cuda_visible"] = environ["CUDA_VISIBLE_DEVICES"]107        return result108    def get_related_packages(self) -> List[str]:109        import pkg_resources110        installed_packages = pkg_resources.working_set111        related_packages = [112            "onnxruntime-gpu",113            "onnxruntime",114            "ort-nightly-gpu",115            "ort-nightly",116            "onnx",117            "transformers",118            "protobuf",119            "sympy",120            "torch",121            "tensorflow",122            "flatbuffers",123            "numpy",124            "onnxconverter-common",125        ]126        related_packages_list = {i.key: i.version for i in installed_packages if i.key in related_packages}127        return related_packages_list128    def get_onnxruntime_info(self) -> Dict:129        try:130            import onnxruntime131            return {132                "version": onnxruntime.__version__,133                "support_gpu": "CUDAExecutionProvider" in onnxruntime.get_available_providers(),134            }135        except ImportError as error:136            if not self.silent:137                self.logger.exception(error)138            return None139        except Exception as exception:140            if not self.silent:141                self.logger.exception(exception, False)142            return None143    def get_pytorch_info(self) -> Dict:144        try:145            import torch146            return {147                "version": torch.__version__,148                "support_gpu": torch.cuda.is_available(),149                "cuda": torch.version.cuda,150            }151        except ImportError as error:152            if not self.silent:153                self.logger.exception(error)154            return None155        except Exception as exception:156            if not self.silent:157                self.logger.exception(exception, False)158            return None159    def get_tensorflow_info(self) -> Dict:160        try:161            import tensorflow as tf162            return {163                "version": tf.version.VERSION,164                "git_version": tf.version.GIT_VERSION,165                "support_gpu": tf.test.is_built_with_cuda(),166            }167        except ImportError as error:168            if not self.silent:169                self.logger.exception(error)170            return None171        except ModuleNotFoundError as error:172            if not self.silent:173                self.logger.exception(error)174            return None175def parse_arguments():176    parser = argparse.ArgumentParser()177    parser.add_argument(178        "--silent",179        required=False,180        action="store_true",181        help="Do not print error message",182    )183    parser.set_defaults(silent=False)184    args = parser.parse_args()185    return args186def get_machine_info(silent=True) -> str:187    machine = MachineInfo(silent)188    return json.dumps(machine.machine_info, indent=2)189if __name__ == "__main__":190    args = parse_arguments()...

cost.py

Source:cost.py

...9MONTH_HOURS = 73010def get_pricelist():11    response = requests.get(PRICELIST)12    return response.json()['gcp_price_list']13def get_machine_info(call):14  jes = call['jes']15  zone = jes['zone']16  region = zone[:-2]17  machine_type = jes['machineType']18  if machine_type.startswith(zone):19    machine_type = machine_type.split('/')[1]20  return region, machine_type, call['preemptible']21def get_price_key(key, preemptible):22  return key + ('-PREEMPTIBLE' if preemptible else '')23def get_machine_hour(pricelist, region, machine_type, preemptible = False):24  price = 025  if machine_type.startswith('custom'):26    _, core, memory = machine_type.split('-')27    core_key = get_price_key('CP-COMPUTEENGINE-CUSTOM-VM-CORE', preemptible)28    price += pricelist[core_key][region] * int(core)29    memory_key = get_price_key('CP-COMPUTEENGINE-CUSTOM-VM-RAM', preemptible)30    price += pricelist[memory_key][region] * ceil(int(memory) * 0.001)31  else:32    price_key = get_price_key(33      'CP-COMPUTEENGINE-VMIMAGE-' + machine_type.upper(), preemptible34    )35    price += pricelist[price_key][region] * memory * 0.00136  return price37def get_disk_hour(call, pricelist):38  region, _, preemptible = get_machine_info(call)39  disks = call['runtimeAttributes']['disks'].split(',')40  for disk in disks:41    _, disk_size, disk_type = disk.strip().split()42    price_key = 'CP-COMPUTEENGINE-'43    if disk_type == 'HDD':44      price_key += 'STORAGE-PD-CAPACITY'45    elif disk_type == 'SSD':46      price_key += 'STORAGE-PD-SSD'47    elif disk_type == 'LOCAL':48      disk_size = '375'49      price_key += 'LOCAL-SSD'50      if preemptible:51        price_key += '-PREEMPTIBLE'52  return pricelist[price_key][region] * int(disk_size) / MONTH_HOURS53def get_datetime(dt):54  return dateutil.parser.parse(dt)55def get_hours(call):56  start_time = get_datetime(call['start'])57  if 'end' in call:58      end_time = get_datetime(call['end'])59  else:60      end_time = datetime.now(timezone.utc)61  delta = end_time - start_time62  seconds = delta.days * 24 * 3600 + delta.seconds63  return max(seconds, 60) / 3600.064def get_price(metadata, pricelist, price=0):65  for calls in metadata['calls'].values():66    for call in calls:67      if 'jes' in call and 'zone' in call['jes'] and 'machineType' in call['jes']:68        machine_hour = get_machine_hour(pricelist, *get_machine_info(call))69        disk_hour = get_disk_hour(call, pricelist)70        hours = get_hours(call)71        price += (machine_hour + disk_hour) * hours72      if "subWorkflowMetadata" in call:73        price += get_price(call["subWorkflowMetadata"], pricelist, price)74  return ceil(price * 100) / 100.075def main():76  metadata = json.loads(sys.stdin.read())77  pricelist = get_pricelist()78  price = get_price(metadata, pricelist, price=0)79  print('$%.2f' % price)80if __name__ == '__main__':...

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.