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

How to use start_forwarding method in lisa

Best Python code snippet using lisa_python

server.py

Source:server.py

...55            if data == b"":56                break57            writer.write(data)58            await writer.drain()59async def start_forwarding(60    reader: StreamReader,61    writer: StreamWriter,62    remote_reader: StreamReader,63    remote_writer: StreamWriter,64) -> None:65    with closing(remote_writer):66        writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")67        await writer.drain()68        await asyncio.gather(69            forward_data(remote_writer, reader),70            forward_data(writer, remote_reader),71        )72async def handle_request(reader: StreamReader, writer: StreamWriter) -> None:73    data = await reader.readuntil(b"\r\n\r\n")74    config = parse_request(data)75    if config:76        try:77            remote_reader, remote_writer = await asyncio.open_connection(78                config.host, config.port79            )80            await start_forwarding(reader, writer, remote_reader, remote_writer)81            await asyncio.to_thread(82                add_visit_stat,83                SQL_INSERT_WEB_STAT.format(url=config.host, time=datetime.now()),84            )85        except (86            ConnectionRefusedError,87            ConnectionResetError,88            socket.gaierror,89            TimeoutError,90        ) as e:91            logger.warning(e.strerror)92    else:93        writer.write(b"Blocked \r\n\r\n")94        await writer.drain()...

main.py

Source:main.py

...7count_msgs = 08last_msg_text = ''9proc = 010session = 'inactive'11def start_forwarding(client):12	print("-> Starting forwarder...")13	chnl = client.get_entity(from_channel_link)14	target = client.get_entity(to_user)15	last_msg_id = 016	print("-> Forwarding messages...")17	session = 'active'18	while True:19		msgs = client.get_messages(chnl)20		new_msg = msgs[0]21		if new_msg.id != last_msg_id:22			last_msg_id = new_msg.id23			client.forward_messages(target, new_msg)24			print('New channel message forwarded:')25			print(new_msg.message)...

camera.py

Source:camera.py

1import numpy as np2import cv23import time4from Motor import *            5PWM=Motor() 6# Video source - can be camera index number given by 'ls /dev/video*7# or can be a video file, e.g. '~/Video.avi'8cap = cv2.VideoCapture(0)9start_moving = 010start_forwarding = 011while(True):12    # Capture frame-by-frame13    ret, frame = cap.read()14    # Our operations on the frame come here15    #gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)16    17    # Filter18    hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)19    lower_red = np.array([0, 100, 100])20    upper_red = np.array([10, 255, 255])21    mask = cv2.inRange(hsv, lower_red, upper_red)22    result = cv2.bitwise_and(frame, frame, mask = mask)       23    24    gray = cv2.cvtColor(result, cv2.COLOR_BGR2GRAY)25    gray = cv2.medianBlur(gray, 5) 26    27    rows = gray.shape[0]28    circles = cv2.HoughCircles(gray, cv2.HOUGH_GRADIENT, 1, rows / 8,29                               param1=100, param2=15,30                               minRadius=1, maxRadius=200)31    32    if circles is None:33        start_moving = start_moving +134    35    if start_moving > 5:36        PWM.setMotorModel(950,950,-950,-950)37    if circles is not None:38        start_moving = 039        PWM.setMotorModel(0,0,0,0)40        circles = np.uint16(np.around(circles))41        biggest = circles[0, 0]42        for i in circles[0, :]:43            if i[2] > biggest[2] :44                biggest = i45                46        center = (biggest[0], biggest[1])47        # circle center48        cv2.circle(gray, center, 1, (0, 100, 100), 3)49        # circle outline50        radius = biggest[2]51        cv2.circle(gray, center, radius, (255, 0, 255), 3)52        print(radius, center)53        54        if radius < 80 :55            start_forwarding = start_forwarding - 156        else:57            start_forwarding = -158        59        if start_forwarding >= 3:60            speed = -60061            PWM.setMotorModel(speed,speed,speed,speed)62        elif start_forwarding <= -3:63            speed = 60064            PWM.setMotorModel(speed,speed,speed,speed)65        else :66            PWM.setMotorModel(0,0,0,0)67            68    # Display the resulting frame69    cv2.imshow('frame',gray)70    71    if cv2.waitKey(1) & 0xFF == ord('q'):72        PWM.setMotorModel(0,0,0,0)73        break74# When everything done, release the capture75cap.release()...

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.