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

How to use _on_frame_received method in Playwright Python

Best Python code snippet using playwright-python

_network.py

Source:_network.py

...238            lambda params: self._on_frame_sent(params["opcode"], params["data"]),239        )240        self._channel.on(241            "frameReceived",242            lambda params: self._on_frame_received(params["opcode"], params["data"]),243        )244        self._channel.on(245            "error", lambda params: self.emit(WebSocket.Events.Error, params["error"])246        )247        self._channel.on("close", lambda params: self._on_close())248    @property249    def url(self) -> str:250        return self._initializer["url"]251    def expect_event(252        self,253        event: str,254        predicate: Callable = None,255        timeout: float = None,256    ) -> EventContextManagerImpl:257        if timeout is None:258            timeout = cast(Any, self._parent)._timeout_settings.timeout()259        wait_helper = WaitHelper(self._loop)260        wait_helper.reject_on_timeout(261            timeout, f'Timeout while waiting for event "{event}"'262        )263        if event != WebSocket.Events.Close:264            wait_helper.reject_on_event(265                self, WebSocket.Events.Close, Error("Socket closed")266            )267        if event != WebSocket.Events.Error:268            wait_helper.reject_on_event(269                self, WebSocket.Events.Error, Error("Socket error")270            )271        wait_helper.reject_on_event(self._parent, "close", Error("Page closed"))272        wait_helper.wait_for_event(self, event, predicate)273        return EventContextManagerImpl(wait_helper.result())274    async def wait_for_event(275        self, event: str, predicate: Callable = None, timeout: float = None276    ) -> Any:277        async with self.expect_event(event, predicate, timeout) as event_info:278            pass279        return await event_info280    def _on_frame_sent(self, opcode: int, data: str) -> None:281        if opcode == 2:282            self.emit(WebSocket.Events.FrameSent, base64.b64decode(data))283        else:284            self.emit(WebSocket.Events.FrameSent, data)285    def _on_frame_received(self, opcode: int, data: str) -> None:286        if opcode == 2:287            self.emit(WebSocket.Events.FrameReceived, base64.b64decode(data))288        else:289            self.emit(WebSocket.Events.FrameReceived, data)290    def is_closed(self) -> bool:291        return self._is_closed292    def _on_close(self) -> None:293        self._is_closed = True294        self.emit(WebSocket.Events.Close)295def serialize_headers(headers: Dict[str, str]) -> List[Header]:296    return [{"name": name, "value": value} for name, values in headers.items() for value in ([str(value) for value in values] if type(values) in [list,set,tuple] else [str(values)])]297def parse_headers(headers: List[Header]) -> Dict[str, str]:...

connection.py

Source:connection.py

...155        raise_error_from_server(frame.reply_code, frame.reply_text)156    async def _handle_server_heartbeat(self, frame: HeartbeatFrame):157        print("Server Heartbeat Received")158        self.missed_heartbeats = 0159    async def _on_frame_received(self, frame: Frame):160        try:161            await self.router.route_frame(frame)162        except KeyError:163            print("Frame {} has no router. Skipping it.".format(frame))164    async def _heartbeat_loop(self):165        print("Initiating Heartbeat Loop")166        while self._running:167            await asyncio.sleep(self.heartbeat // 2)168            seconds_without_send = (169                    datetime.utcnow() - self.last_send_dt170            ).total_seconds()171            if seconds_without_send > self.heartbeat:172                heartbeat = Frame.from_frame(HeartbeatFrame(), channel=0)173                await self._send_to_server(heartbeat)174    async def _run(self):175        """176        Read Frames until the connection closes.177        :return:178        """179        self._running = True180        while self._running:181            try:182                header = await asyncio.wait_for(183                    self.conn.recv(7), timeout=self.heartbeat184                )185            except asyncio.TimeoutError:186                print("Read Timed-out increasing missed heartbeats count")187                self.missed_heartbeats += 1188                if self.missed_heartbeats > 4:189                    await self.close()190                    raise ConnectionAbortedError(191                        "Server missed heartbeats. Closing connection"192                    )193                continue194            if header.startswith(b"AMQP"):195                # Some Protocol Version Error.196                header += await self.conn.recv(1)197                supported_version = struct.unpack(">BBB", header[-3:])198                raise ProtocolError(199                    f"Target server does not support {VERSION} version. "200                    f"Supported Version is: {supported_version}"201                )202            frame, remaining = await self.parse_frame(header)203            assert remaining == b'', 'Remaining of {} is not empty: {}'.format(204                frame, remaining205            )206            # remaining should always be empty207            await self._on_frame_received(frame)208    async def parse_frame(self, header) -> Tuple[Frame, bytes]:209        frame_header, remaining = FrameHeader.from_bytes(header)210        payload = await self.conn.recv(frame_header.size + 1)211        if not payload.endswith(FRAME_END):212            raise FrameEndError(213                "Expected Frame-End not found after reading the whole "214                "frame data."215            )216        payload = payload.rstrip(FRAME_END)217        return Frame.from_bytes(header + payload)218    async def close(self):219        self._running = False220async def _get_connection(221        host: str, port: int, ssl=None, loop=None...

client.py

Source:client.py

...57    @property58    def on_frame_received(self) -> EventListener[OnFrameReceivedCallback]:59        """Event triggered when a frame is received by the client."""60        return self._on_frame_received_event61    def _on_frame_received(self, frame_index: int, frame: FrameData) -> None:62        changes = frame.array_keys | frame.value_keys63        super()._on_frame_received(frame_index, frame)64        self._on_frame_received_event.invoke(changes=changes)65    @classmethod66    @contextmanager67    def connect_to_session(cls, session: Session, /) -> Generator[Client, None, None]:68        """69        Connect to a session.70        :param session: Session that is running.71        :yields: Client connected to the given session.72        """73        with cls.connect_to_server(address="localhost", port=session.port) as client:74            yield client75    @classmethod76    @contextmanager77    def connect_to_server(...

yolov4_dices_detection_controller.py

Source:yolov4_dices_detection_controller.py

...40        self._streamer.moveToThread(self._thread)41        self._streamer.frames.connect(self._on_frame_received)42        self._thread.started.connect(self._streamer.run)43        self._thread.start()44    def _on_frame_received(self, frame: np.ndarray):45        dices = self._detector.detect(frame)46        for dice in dices:47            box = dice.box48            left, top = box.left, box.top49            right, bottom = box.left + box.width, box.top + box.height50            color = (0, 0, 255)51            frame = cv2.rectangle(frame, (left, top), (right, bottom), color)52            class_label_x = left + box.width // 253            class_label_y = top + 1054            class_label = _classes[dice.class_id]55            frame = _place_class_label(frame, class_label_x, class_label_y, class_label)56        self.images.emit(rgb_image_to_qt(frame))57    def clear_resources(self):58        self._thread.stop()

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.