KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _collect_message method in localstack

Best Python code snippet using localstack_python

dataclasses.py

Source:dataclasses.py

...70    def wait_running(self):71        return self._state_changed(JobState.RUNNING)72    def wait_done(self):73        return self._state_changed(JobState.DONE)74    def _collect_message(self, result, was_collected, msg):75        if not was_collected.cancelled():76            was_collected.set_result(True)77        if not result.cancelled():78            result.set_result(msg)79    def collect_message(self):80        result = self.loop.create_future()81        if self._msg_inbox:82            self._collect_message(result, *self._msg_inbox.popleft())83        else:84            self._msg_waiters.append(result)85        return result86    def enqueue_message(self, msg):87        was_collected = self.loop.create_future()88        if self._msg_waiters:89            self._collect_message(90                self._msg_waiters.popleft(), was_collected, msg91            )92        else:93            self._msg_inbox.append((was_collected, msg))94        return was_collected95@dataclass96class GroupConfig:97    ident: str = None98    max_jobs: int = 099    max_cpu: float = 0.0100class Group(dict):101    def __init__(self, loop, stats, config):102        self.loop = loop103        self.stats = stats...

gpu_monitor.py

Source:gpu_monitor.py

...36        return (sum(current_gpu_status.occupied_memory) - sum(previous_gpu_status.occupied_memory)) / previous_gpu_status.total_memory37    memory_usage_change = tuple(relative_memory_change(p_gpu, c_gpu) for p_gpu, c_gpu in zip(previous_status, current_status))38    considerable_memory_usage_change = any(abs(change) > 0.5 for change in memory_usage_change)39    return considerable_memory_usage_change or current_total_memory != previous_total_memory or current_users != previous_users40def _collect_message(status: Tuple[GPUStatus, ...]) -> str:41    username_descriptions = {pw.pw_name: pw.pw_gecos for pw in getpwall()}42    msg = 'Status changed!'43    for gpu_idx, gpu_status in enumerate(status):44        utilization = (sum(gpu_status.occupied_memory) / gpu_status.total_memory) * 10045        msg += f'\n\n<b>GPU{gpu_idx}</b> <code>{sum(gpu_status.occupied_memory)}/{gpu_status.total_memory}</code> Mb ({utilization:.2f}%):'46        if not len(gpu_status.occupied_by):47            msg += '\n    vacant!'48        for username, memory in zip(gpu_status.occupied_by, gpu_status.occupied_memory):49            description = username_descriptions[username].replace(',,,', '')50            if len(description):51                msg += f'\n    {username} ({description}), <code>{memory}</code> Mb'52            else:53                msg += f'\n    {username}, <code>{memory}</code> Mb'54    return msg55def run_monitoring(logger: Callable[[str], None], interval: int = 1) -> None:56    previous_status = tuple()57    while True:58        try:59            current_status = _collect_gpu_status()60            if _status_considerably_changed(previous_status, current_status):61                msg = _collect_message(current_status)62                logger(msg)63                previous_status = current_status64        except Exception as e:65            logging.debug('Exception occurred during GPU status evaluation!', exc_info=e)...

s3.py

Source:s3.py

...37    def create_command(self, command: int) -> bytearray:38        command_special = 1 if command == self.command_PAIR else 039        return bytearray([self.command_prefix, command, command_special, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,40                          self.command_suffix])41    def _collect_message(self, package: bytearray) -> bool:42        self._data = package43        return True44    def _decode_response(self, response: bytearray):45        _LOGGER.debug("Data is %s", bytes(response).hex())46        try:47            self._fan_speed = int(list("{:02x}".format(response[2]))[1])48            self._mode = int(list("{:02x}".format(response[2]))[0])49            self._heater = response[4] & 150            self._state = response[4] >> 1 & 151            self._heater_temp = response[3]52            self._sound = response[4] >> 3 & 153            self._out_temp = self.decode_temperature(response[7])54            self._in_temp = self.decode_temperature(response[8])55            self._filter_remain = response[10] * 256 + response[9]...

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.