How to use start_async method in lisa

Best Python code snippet using lisa_python

test_websocket.py

Source:test_websocket.py

...30    """"Open then immediately close"""31    def on_open():32        print('opened')33        client.close()34    await client.start_async(on_open=on_open, on_error=on_error)35    print('closed')36@pytest.mark.asyncio37async def test_trades(client, symbols):38    print(symbols)39    results = []40    def callback(msg):41        results.append(msg)42        client.close()43    def on_open():44        client.subscribe_trades(symbols=symbols, callback=callback)45    await client.start_async(on_open=on_open, on_error=on_error)46    result = results[0]47    assert result['stream'] == 'trades'48@pytest.mark.asyncio49async def test_market_diff(client, symbols):50    results = []51    def callback(msg):52        results.append(msg)53        client.close()54    def on_open():55        client.subscribe_market_diff(symbols=symbols, callback=callback)56    await client.start_async(on_open=on_open, on_error=on_error)57    result = results[0]58    assert result['stream'] == 'marketDiff'59@pytest.mark.asyncio60async def test_market_depth(client, symbols):61    results = []62    def callback(msg):63        results.append(msg)64        client.close()65    def on_open():66        client.subscribe_market_depth(symbols=symbols, callback=callback)67    await client.start_async(on_open=on_open, on_error=on_error)68    result = results[0]69    assert result['stream'] == 'marketDepth'70@pytest.mark.asyncio71async def test_kline(client, symbols):72    results = []73    def callback(msg):74        results.append(msg)75        client.close()76    def on_open():77        client.subscribe_kline(interval='1m', symbols=symbols, callback=callback)78    await client.start_async(on_open=on_open, on_error=on_error)79    result = results[0]80    assert result['stream'] == 'kline_1m'81@pytest.mark.asyncio82async def test_tickers(client, symbols):83    results = []84    def callback(msg):85        results.append(msg)86        client.close()87    def on_open():88        client.subscribe_ticker(symbols=symbols, callback=callback)89    await client.start_async(on_open=on_open, on_error=on_error)90    result = results[0]91    assert result['stream'] == 'ticker'92@pytest.mark.asyncio93async def test_all_tickers(client):94    results = []95    def callback(msg):96        results.append(msg)97        client.close()98    def on_open():99        client.subscribe_all_tickers(callback=callback)100    await client.start_async(on_open=on_open, on_error=on_error)101    result = results[0]102    assert result['stream'] == 'allTickers'103@pytest.mark.asyncio104async def test_mini_ticker(client, symbols):105    results = []106    def callback(msg):107        results.append(msg)108        client.close()109    def on_open():110        client.subscribe_mini_ticker(symbols=symbols, callback=callback)111    await client.start_async(on_open=on_open, on_error=on_error)112    result = results[0]113    assert result['stream'] == 'miniTicker'114@pytest.mark.asyncio115async def test_all_mini_ticker(client, symbols):116    results = []117    def callback(msg):118        results.append(msg)119        client.close()120    def on_open():121        client.subscribe_all_mini_tickers(callback=callback)122    await client.start_async(on_open=on_open, on_error=on_error)123    result = results[0]124    assert result['stream'] == 'allMiniTickers'125@pytest.mark.asyncio126async def test_blockheight(client):127    results = []128    def callback(msg):129        results.append(msg)130        client.close()131    def on_open():132        client.subscribe_blockheight(callback=callback)133    await client.start_async(on_open=on_open, on_error=on_error)134    result = results[0]135    assert 'stream' in result136@pytest.mark.asyncio137async def test_keepalive(client):138    def on_open():139        client.keepalive()140        client.close()141    await client.start_async(on_open=on_open, on_error=on_error)142@pytest.mark.asyncio143async def test_unsubscribe(client):144    results = []145    def callback(msg):146        results.append(msg)147        client.unsubscribe("blockheight")148        client.close()149    def on_open():150        client.subscribe_blockheight(callback=callback)151    await client.start_async(on_open=on_open, on_error=on_error)152    assert results153@pytest.mark.asyncio154async def test_decorator(client):155    @client.on('open')156    def callback():157        client.close()158    await client.start_async()159@pytest.mark.asyncio160async def test_decorator_async(client):161    @client.on('open')162    async def callback():163        client.close()164    await client.start_async()165@pytest.mark.asyncio166async def test_decorator_sub_queue(client):167    results = []168    @client.on("allTickers", symbols=["$all"])169    async def callback(msg):170        results.append(msg)171        client.close()172    await client.start_async()...

reid.py

Source:reid.py

...12            plt.imshow(img)13            plt.show()14            print("ffff")15            img = preprocess(img,size)16            infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})17            status=infer_res.wait()18            results = exec_net.requests[0].outputs[output_layer][0]19            for j,bbox1 in ids1.items():20                print("gggg")21                img1 = frame1[bbox1[1]:bbox1[3], bbox1[0]:bbox1[2]]22                #plt.imshow(img1)23                #plt.show()24                img1 = preprocess(img1,size)25                infer_res = exec_net.start_async(request_id=1,inputs={input_layer:img1})26                status=infer_res.wait()27                results1 = exec_net.requests[1].outputs[output_layer][0]28                print(results1.shape)29                x = torch.tensor(results).unsqueeze(0)30                y = torch.tensor(results1).unsqueeze(0)31                dis=torch.cosine_similarity(x, y)[0].numpy()32                if dis >= 0.5:33                    track_id[i]= bbox34                35def reidentification(ids,track_id,frame):36    img=frame37    track_id_copy = track_id.copy()38    for i,bbox in ids.items():39        if len(track_id) == 0:40            track_id[i] = bbox41        else:42            img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]43            #plt.imshow(img)44            #plt.show()45            img = preprocess(img,size)46            infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})47            status=infer_res.wait()48            results = exec_net.requests[0].outputs[output_layer][0]49            dis = []50            for j,bbox_t in track_id_copy.items():51                img1 = frame[bbox_t[1]:bbox_t[3], bbox_t[0]:bbox_t[2]]52                img1 = preprocess(img1,size)53                infer_res = exec_net.start_async(request_id=1,inputs={input_layer:img1})54                status=infer_res.wait()55                results1 = exec_net.requests[1].outputs[output_layer][0]56                #print(results1.shape)57                x = torch.tensor(results).unsqueeze(0)58                y = torch.tensor(results1).unsqueeze(0)59                d=torch.cosine_similarity(x, y)[0].numpy()60                dis.append(d)61            try:62                f_idx = dis.index(max(dis))63                f = dis[f_idx]64                if f >= 0.7:65                    print("hhh")66                    track_id[f_idx] = bbox67                    i = f_idx68                else:69                    track_id[i] = bbox70                    i=i71            except Exception as e:72                print(e)73        cv2.putText(frame, str(i), bbox[:2], cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)74        cv2.rectangle(img=frame, pt1=bbox[:2], pt2=bbox[2:], color=(0,255,0), thickness=3)75    return track_id,frame76            77            78def emb(ids,frame):79    r= []80    b=[]81    for i,bbox in ids.items():82        img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]83        img = preprocess(img,size)84        infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})85        status=infer_res.wait()86        results = exec_net.requests[0].outputs[output_layer][0]87        #print(type(results))88        r.append(results)89        b.append(bbox)90       91    return r,b92def ids_feature_(ids,frame):93    ids_feat={}94    for i,bbox in ids.items():95        img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]96        img = preprocess(img,size)97        infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})98        status=infer_res.wait()99        results = exec_net.requests[0].outputs[output_layer][0]100        ids_feat[i]=results101    return ids_feat102def distance_(feature,feat1):103    x = torch.tensor(feature).unsqueeze(0)104    y = torch.tensor(feat1).unsqueeze(0)105    d=torch.cosine_similarity(x, y)[0].numpy()106    return d107def distance_list(feature,feat1):108    x = torch.tensor(feature[0]).unsqueeze(0)109    dis=[]110    for i in feat1:111        y = torch.tensor(i).unsqueeze(0)112        d=torch.cosine_similarity(x, y)[0].numpy()113        dis.append(d)114    115    return sum(dis)/len(dis)116def ids_feature_list(ids,frame):117    ids_feat={}118    for i,bbox in ids.items():119        img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]120        img = preprocess(img,size)121        infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})122        status=infer_res.wait()123        results = exec_net.requests[0].outputs[output_layer][0]124        ids_feat[i]=[results]...

launcher_test.py

Source:launcher_test.py

...10    @mock.patch.object(launcher.update.status, 'get')11    def test_launches_update_when_no_update_is_running(self, mock_status_get,12                                                       mock_clear, mock_popen):13        mock_status_get.return_value = (update_status.Status.NOT_RUNNING, '')14        launcher.start_async()15        mock_clear.assert_called()16        mock_popen.assert_called_once_with(17            ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))18    @mock.patch.object(launcher.subprocess, 'Popen')19    @mock.patch.object(launcher.update.result_store, 'clear')20    @mock.patch.object(launcher.update.status, 'get')21    def test_launches_update_when_previous_update_succeeded(22            self, mock_status_get, mock_clear, mock_popen):23        mock_status_get.return_value = (update_status.Status.DONE, '')24        launcher.start_async()25        mock_clear.assert_called()26        mock_popen.assert_called_once_with(27            ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))28    @mock.patch.object(launcher.subprocess, 'Popen')29    @mock.patch.object(launcher.update.result_store, 'clear')30    @mock.patch.object(launcher.update.status, 'get')31    def test_launches_update_when_previous_update_failed(32            self, mock_status_get, mock_clear, mock_popen):33        mock_status_get.return_value = (update_status.Status.DONE,34                                        'dummy updater failure message')35        launcher.start_async()36        mock_clear.assert_called_once()37        mock_popen.assert_called_once_with(38            ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))39    @mock.patch.object(launcher.subprocess, 'Popen')40    @mock.patch.object(launcher.update.result_store, 'clear')41    @mock.patch.object(launcher.update.status, 'get')42    def test_does_not_launch_if_update_is_already_running(43            self, mock_status_get, mock_clear, mock_popen):44        mock_status_get.return_value = (update_status.Status.IN_PROGRESS, '')45        with self.assertRaises(launcher.AlreadyInProgressError):46            launcher.start_async()47        mock_clear.assert_not_called()...

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.