KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use handle_frame method in ATX
Best Python code snippet using ATX
test_devices.py
Source:test_devices.py 
...57 Response(data={DATA_BOILER_SENSORS: {"test_sensor": 42}}),58 Response(data={DATA_MODE: 1}),59 )60 for frame in frames:61 ecomax.handle_frame(frame)62 assert await ecomax.get_value("test_sensor") == 4263 boiler_control = await ecomax.get_parameter("boiler_control")64 assert isinstance(boiler_control, BoilerBinaryParameter)65 assert boiler_control.value == 166async def test_boiler_parameters_callbacks(ecomax: EcoMAX) -> None:67 """Test callbacks that are fired on received parameter frames."""68 ecomax.handle_frame(69 Response(70 data={71 DATA_BOILER_PARAMETERS: {72 "test_binary_parameter": [0, 0, 1],73 "test_parameter": [10, 5, 20],74 }75 }76 )77 )78 assert await ecomax.get_value("test_binary_parameter") == 079 test_binary_parameter = await ecomax.get_parameter("test_binary_parameter")80 assert isinstance(test_binary_parameter, BoilerBinaryParameter)81 assert await ecomax.get_value("test_parameter") == 1082 test_parameter = await ecomax.get_parameter("test_parameter")83 assert test_parameter.value == 1084 assert test_parameter.min_value == 585 assert test_parameter.max_value == 2086async def test_fuel_consumption_callbacks() -> None:87 """Test callbacks that are fired on received fuel consumption."""88 with patch("time.time", side_effect=(10, 20)):89 ecomax = EcoMAX(asyncio.Queue())90 ecomax.handle_frame(Response(data={DATA_FUEL_CONSUMPTION: 3.6}))91 await ecomax.wait_for_tasks()92 assert await ecomax.get_value("fuel_burned") == 0.0193async def test_regdata_callbacks(94 ecomax: EcoMAX, data_schema: DataSchema, regulator_data: RegulatorData95) -> None:96 """Test callbacks that are fired on received regdata."""97 # Test exception handling on data schema timeout.98 with patch(99 "pyplumio.devices.AsyncDevice.get_value", side_effect=asyncio.TimeoutError100 ):101 ecomax.handle_frame(regulator_data)102 await ecomax.wait_for_tasks()103 # Regulator data should be empty on schema timeout.104 assert not await ecomax.get_value(DATA_REGDATA)105 # Set data schema and parse the regdata.106 ecomax.handle_frame(data_schema)107 ecomax.handle_frame(regulator_data)108 await ecomax.wait_for_tasks()109 regdata = await ecomax.get_value(DATA_REGDATA)110 assert regdata["mode"] == 0111 assert round(regdata["heating_temp"], 1) == 22.4112 assert regdata["heating_target"] == 41113 assert regdata["183"] == "0.0.0.0"114 assert regdata["184"] == "255.255.255.0"115async def test_mixer_sensors_callbacks(ecomax: EcoMAX) -> None:116 """Test callbacks that are fired on receiving mixer sensors info."""117 ecomax.handle_frame(Response(data={DATA_MIXER_SENSORS: [{"test_sensor": 42}]}))118 mixers = await ecomax.get_value("mixers")119 assert len(mixers) == 1120 assert isinstance(mixers[0], Mixer)121 assert mixers[0].index == 0122 assert await mixers[0].get_value("test_sensor") == 42123async def test_mixer_parameters_callbacks(ecomax: EcoMAX) -> None:124 """Test callbacks that are fired on receiving mixer parameters."""125 ecomax.handle_frame(126 Response(127 data={128 DATA_MIXER_PARAMETERS: [129 {130 "test_binary_parameter": [0, 0, 1],131 "test_parameter": [10, 5, 20],132 }133 ]134 }135 )136 )137 mixers = await ecomax.get_value("mixers")138 test_binary_parameter = await mixers[0].get_parameter("test_binary_parameter")139 assert test_binary_parameter.value == 0140 assert isinstance(test_binary_parameter, MixerBinaryParameter)141 test_parameter = await mixers[0].get_parameter("test_parameter")142 assert isinstance(test_parameter, MixerParameter)143 assert test_parameter.value == 10144 assert test_parameter.min_value == 5145 assert test_parameter.max_value == 20146async def test_register_callback(ecomax: EcoMAX) -> None:147 """Test callback registration."""148 mock_callback = AsyncMock(return_value=None)149 ecomax.register_callback("test_sensor", mock_callback)150 ecomax.handle_frame(Response(data={DATA_BOILER_SENSORS: {"test_sensor": 42.1}}))151 await ecomax.wait_for_tasks()152 mock_callback.assert_awaited_once_with(42.1)153 mock_callback.reset_mock()154 # Test with change.155 ecomax.handle_frame(Response(data={DATA_BOILER_SENSORS: {"test_sensor": 45}}))156 await ecomax.wait_for_tasks()157 mock_callback.assert_awaited_once_with(45)158 mock_callback.reset_mock()159 # Remove the callback and make sure it doesn't fire again.160 ecomax.remove_callback("test_sensor", mock_callback)161 ecomax.handle_frame(Response(data={DATA_BOILER_SENSORS: {"test_sensor": 50}}))162 await ecomax.wait_for_tasks()163 mock_callback.assert_not_awaited()164async def test_get_value(ecomax: EcoMAX) -> None:165 """Test getting value from the device."""166 with pytest.raises(RuntimeError), patch(167 "asyncio.sleep", side_effect=RuntimeError("break loop")168 ) as mock_sleep:169 await ecomax.get_value("test")170 mock_sleep.assert_awaited_once_with(0)171async def test_set_value(ecomax: EcoMAX) -> None:172 """Test setting parameter value via set_value helper."""173 with pytest.raises(RuntimeError), patch(174 "asyncio.sleep", side_effect=RuntimeError("break loop")175 ) as mock_sleep:176 await ecomax.set_value("test_parameter", 1)177 mock_sleep.assert_awaited_once_with(0)178 # Test with valid parameter.179 valid = Mock(spec=Parameter)180 ecomax.__dict__["valid_parameter"] = valid181 await ecomax.set_value("valid_parameter", 1)182 valid.set.assert_called_once_with(1)183 # Test with invalid parameter.184 invalid = Mock()185 ecomax.__dict__["invalid_parameter"] = invalid186 with pytest.raises(ParameterNotFoundError):187 await ecomax.set_value("invalid_parameter", 1)188async def test_get_parameter(ecomax: EcoMAX) -> None:189 """Test getting parameter from device."""190 with pytest.raises(RuntimeError), patch(191 "asyncio.sleep", side_effect=RuntimeError("break loop")192 ) as mock_sleep:193 await ecomax.get_parameter("test")194 mock_sleep.assert_awaited_once_with(0)195 # Test with invalid parameter.196 invalid = Mock()197 ecomax.__dict__["invalid_parameter"] = invalid198 with pytest.raises(ParameterNotFoundError):199 await ecomax.get_parameter("invalid_parameter")200@patch("pyplumio.devices.Mixer.shutdown")201@patch("pyplumio.devices.AsyncDevice.cancel_tasks")202@patch("pyplumio.devices.AsyncDevice.wait_for_tasks")203async def test_shutdown(204 mock_wait_for_tasks, mock_cancel_tasks, mock_shutdown, ecomax: EcoMAX205) -> None:206 """Test device tasks shutdown."""207 ecomax.handle_frame(Response(data={DATA_MIXER_SENSORS: [{"test_sensor": 42}]}))208 await ecomax.shutdown()209 mock_wait_for_tasks.assert_awaited_once()210 mock_cancel_tasks.assert_called_once()...
compute.py
Source:compute.py
1import json2def compute_material_required(width: float, height: float):3 # track4 horizontal_track = width * 2 # 2 pcs5 vertical_track = height * 2 # 2 pcs6 total_track_length = 2 * (width + height) #Unit: mm7 # framing8 horizontal_frame = (width - 9)/2 # 1 piece of horizontal frame9 handle_frame = height - 6910 total_framing_length = 6 * (horizontal_frame + handle_frame) #Unit: mm11 # glass12 glass_area = ((horizontal_frame - 105) *13 (handle_frame - 105)) * (0.00328084**2) #Unit: sq. ft14 # mosquito net15 mosquito_area = (horizontal_frame * handle_frame) * (0.00328084**2) #Unit: sq. ft16 # glass pvc17 glass_pvc_vertical_height = handle_frame * 4 #Unit: mm18 glass_pvc_width = width * 219 # c chanel20 c_chanel_height = handle_frame * 2 #Unit: mm21 c_chanel_width = width22 # track woolpile23 woolpile_height = handle_frame * 9 # TODO: Correction24 woolpile_width = horizontal_frame * 2 # TODO: Correction25 return {26 "Track - Top/Bottom": {27 "per_piece": round(horizontal_track / 2, 3),28 "total": horizontal_track,29 "Unit": "mm",30 "Pieces": "2"31 },32 "Track - Vertical": {33 "per_piece": round(vertical_track / 2, 3),34 "total": vertical_track ,35 "Unit": "mm",36 "Pieces": "2"37 },38 "Section - Top/Bottom" : {39 "per_piece": round(horizontal_frame / 6, 3),40 "total": horizontal_frame,41 "Unit": "mm",42 "Pieces": "6"43 },44 "Section - Handle": {45 "per_piece": round(handle_frame / 6, 3),46 "total": handle_frame,47 "Unit": "mm",48 "Pieces": "6"49 },50 "Glass": {51 "per_piece": round(glass_area/2, 3),52 "total": round(glass_area, 3),53 "Unit": "sq. ft",54 "Pieces": "2"55 },56 "Mosquito Net": {57 "per_piece": round(mosquito_area, 3),58 "total": round(mosquito_area, 3),59 "Unit": "sq. ft",60 "Pieces": "1"61 },62 "Glass PVC - Horizontal": {63 "total": glass_pvc_width,64 "Unit": "mm",65 "Pieces": "4",66 "per_piece": round(glass_pvc_width/4, 3)67 },68 "Glass PVC - Vertical": {69 "total": glass_pvc_vertical_height,70 "Unit": "mm",71 "Pieces": "4",72 "per_piece": round(glass_pvc_vertical_height/4, 3)73 },74 "C Chanel - Horizontal": {75 "Unit": "mm",76 "Pieces": "1",77 "total": c_chanel_width,78 "per_piece": c_chanel_width79 },80 "C Chanel - Vertical": {81 "Unit": "mm",82 "Pieces": "2",83 "total": c_chanel_height,84 "per_piece": round(c_chanel_height/2, 3)85 },86 "Woolpile - Horizontal": {87 "Unit": "mm",88 "Pieces": "2",89 "total":woolpile_width,90 "per_piece": round(woolpile_width / 2, 3)91 },92 "Woolpile - Vertical": {93 "Unit": "mm",94 "Pieces": "9",95 "total":woolpile_height,96 "per_piece": round(woolpile_height / 9, 3)97 },98 "Inter Lock Strip" : {99 "Pieces": "3",100 "Unit":"mm",101 "per_piece":handle_frame,102 "total":handle_frame * 3,103 },104 "Bearing" : {105 "Pieces": "6",106 "Unit":"",107 "per_piece":"-",108 "total":"-",109 },110 "Lock" : {111 "Pieces": "2",112 "Unit":"",113 "per_piece":"-",114 "total":"-",115 },116 "Aluminium Clit" : {117 "Pieces": "20",118 "Unit":"",119 "per_piece":"-",120 "total":"-",121 },122 "Corner Clit" : {123 "Pieces": "24",124 "Unit":"",125 "per_piece":"-",126 "total":"-",127 },128 "Male Female" : {129 "Pieces": "6",130 "Unit":"",131 "per_piece":"-",132 "total":"-",133 },134 "Long Cap" : {135 "Pieces": "3",136 "Unit":"",137 "per_piece":"-",138 "total":"-",139 },140 "Lock Guard" : {141 "Pieces": "3",142 "Unit":"",143 "per_piece":"-",144 "total":"-",145 },146 }...
snake.py
Source:snake.py
...14 dis_key = snake_body.pop(0)15 f.write(f"{dis_key}=000000\n")16# Frame counter for section headings17frame_counter = 118def handle_frame(f: TextIO, key: str):19 global frame_counter20 # Print the section head as well as the21 # snake body delta, followed by a newline for readability22 f.write(f"[{frame_counter}]\n")23 snake_append(f, key)24 f.write("\n")25 frame_counter += 126def generate(f: TextIO):27 # Print the settings section28 f.write("[settings]\n")29 f.write("frame_del={snake_speed}\n")30 f.write("draw_mode=KDM_ADD_ALL\n\n")31 # Create serpentines32 for rowi in range(0, len(keyrows)):33 row = keyrows[rowi]34 rowlen = len(row)35 for coli in range(0, rowlen):36 # Reverse uneven rows37 i = coli if rowi % 2 == 0 else rowlen - coli - 138 handle_frame(f, row[i])39 # Go back to the start40 for rowi in reversed(range(0, len(keyrows) - 1)):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run ATX automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
