Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.