diff --git a/examples/control.py b/examples/control.py index a95ef488..5238e59b 100644 --- a/examples/control.py +++ b/examples/control.py @@ -8,21 +8,20 @@ async def main() -> None: """Show example on controlling your WLED device.""" - async with WLED("10.10.11.61") as led: + async with WLED("10.10.11.31") as led: device = await led.update() print(device.info.version) + print(device.state) - print(device.info.leds) - print(device.state.segments[0]) - # await led.segment( - # 0, - # await led.segment( + if device.state.on: + print("Turning off WLED....") + await led.master(on=False) + else: + print("Turning on WLED....") + await led.master(on=True) - # if isinstance(device.state.preset, Preset): - - # if isinstance(device.state.playlist, Playlist): - - # Turn strip on, full brightness + device = await led.update() + print(device.state) if __name__ == "__main__": diff --git a/examples/upgrade.py b/examples/upgrade.py index 35a3aee9..095050f3 100644 --- a/examples/upgrade.py +++ b/examples/upgrade.py @@ -3,24 +3,31 @@ import asyncio -from wled import WLED +from wled import WLED, WLEDReleases async def main() -> None: """Show example on upgrade your WLED device.""" + async with WLEDReleases() as releases: + latest = await releases.releases() + print(f"Latest stable version: {latest.stable}") + print(f"Latest beta version: {latest.beta}") + + if not latest.stable: + print("No stable version found") + return + async with WLED("10.10.11.54") as led: device = await led.update() - print(f"Latest stable version: {device.info.version_latest_stable}") - print(f"Latest beta version: {device.info.version_latest_beta}") print(f"Current version: {device.info.version}") print("Upgrading WLED....") - await led.upgrade(version="0.13.0-b4") + await led.upgrade(version=latest.stable) print("Waiting for WLED to come back....") await asyncio.sleep(5) - device = await led.update(full_update=True) + device = await led.update() print(f"Current version: {device.info.version}") diff --git a/poetry.lock b/poetry.lock index b1d1e7cf..039401b1 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand. [[package]] name = "aiohttp" @@ -912,6 +912,26 @@ dev = ["marshmallow[tests]", "pre-commit (>=3.5,<4.0)", "tox"] docs = ["alabaster (==0.7.16)", "autodocsumm (==0.2.12)", "sphinx (==7.3.7)", "sphinx-issues (==4.1.0)", "sphinx-version-warning (==1.1.2)"] tests = ["pytest", "pytz", "simplejson"] +[[package]] +name = "mashumaro" +version = "3.13.1" +description = "Fast and well tested serialization library" +optional = false +python-versions = ">=3.8" +files = [ + {file = "mashumaro-3.13.1-py3-none-any.whl", hash = "sha256:ad0a162b8f4ea232dadd2891d77ff20165b855b9d84610f36ac84462d4576aa0"}, + {file = "mashumaro-3.13.1.tar.gz", hash = "sha256:169f0290253b3e6077bcb39c14a9dd0791a3fdedd9e286e536ae561d4ff1975b"}, +] + +[package.dependencies] +typing-extensions = ">=4.1.0" + +[package.extras] +msgpack = ["msgpack (>=0.5.6)"] +orjson = ["orjson"] +toml = ["tomli (>=1.1.0)", "tomli-w (>=1.0)"] +yaml = ["pyyaml (>=3.13)"] + [[package]] name = "mccabe" version = "0.7.0" @@ -1101,6 +1121,61 @@ files = [ {file = "nodeenv-1.9.1.tar.gz", hash = "sha256:6ec12890a2dab7946721edbfbcd91f3319c6ccc9aec47be7c7e6b7011ee6645f"}, ] +[[package]] +name = "orjson" +version = "3.10.5" +description = "Fast, correct Python JSON library supporting dataclasses, datetimes, and numpy" +optional = false +python-versions = ">=3.8" +files = [ + {file = "orjson-3.10.5-cp310-cp310-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:545d493c1f560d5ccfc134803ceb8955a14c3fcb47bbb4b2fee0232646d0b932"}, + {file = "orjson-3.10.5-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f4324929c2dd917598212bfd554757feca3e5e0fa60da08be11b4aa8b90013c1"}, + {file = "orjson-3.10.5-cp310-cp310-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:8c13ca5e2ddded0ce6a927ea5a9f27cae77eee4c75547b4297252cb20c4d30e6"}, + {file = "orjson-3.10.5-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b6c8e30adfa52c025f042a87f450a6b9ea29649d828e0fec4858ed5e6caecf63"}, + {file = "orjson-3.10.5-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:338fd4f071b242f26e9ca802f443edc588fa4ab60bfa81f38beaedf42eda226c"}, + {file = "orjson-3.10.5-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:6970ed7a3126cfed873c5d21ece1cd5d6f83ca6c9afb71bbae21a0b034588d96"}, + {file = "orjson-3.10.5-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:235dadefb793ad12f7fa11e98a480db1f7c6469ff9e3da5e73c7809c700d746b"}, + {file = "orjson-3.10.5-cp310-none-win32.whl", hash = "sha256:be79e2393679eda6a590638abda16d167754393f5d0850dcbca2d0c3735cebe2"}, + {file = "orjson-3.10.5-cp310-none-win_amd64.whl", hash = "sha256:c4a65310ccb5c9910c47b078ba78e2787cb3878cdded1702ac3d0da71ddc5228"}, + {file = "orjson-3.10.5-cp311-cp311-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:cdf7365063e80899ae3a697def1277c17a7df7ccfc979990a403dfe77bb54d40"}, + {file = "orjson-3.10.5-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6b68742c469745d0e6ca5724506858f75e2f1e5b59a4315861f9e2b1df77775a"}, + {file = "orjson-3.10.5-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:7d10cc1b594951522e35a3463da19e899abe6ca95f3c84c69e9e901e0bd93d38"}, + {file = "orjson-3.10.5-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dcbe82b35d1ac43b0d84072408330fd3295c2896973112d495e7234f7e3da2e1"}, + {file = "orjson-3.10.5-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:10c0eb7e0c75e1e486c7563fe231b40fdd658a035ae125c6ba651ca3b07936f5"}, + {file = "orjson-3.10.5-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:53ed1c879b10de56f35daf06dbc4a0d9a5db98f6ee853c2dbd3ee9d13e6f302f"}, + {file = "orjson-3.10.5-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:099e81a5975237fda3100f918839af95f42f981447ba8f47adb7b6a3cdb078fa"}, + {file = "orjson-3.10.5-cp311-none-win32.whl", hash = "sha256:1146bf85ea37ac421594107195db8bc77104f74bc83e8ee21a2e58596bfb2f04"}, + {file = "orjson-3.10.5-cp311-none-win_amd64.whl", hash = "sha256:36a10f43c5f3a55c2f680efe07aa93ef4a342d2960dd2b1b7ea2dd764fe4a37c"}, + {file = "orjson-3.10.5-cp312-cp312-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:68f85ecae7af14a585a563ac741b0547a3f291de81cd1e20903e79f25170458f"}, + {file = "orjson-3.10.5-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:28afa96f496474ce60d3340fe8d9a263aa93ea01201cd2bad844c45cd21f5268"}, + {file = "orjson-3.10.5-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:9cd684927af3e11b6e754df80b9ffafd9fb6adcaa9d3e8fdd5891be5a5cad51e"}, + {file = "orjson-3.10.5-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3d21b9983da032505f7050795e98b5d9eee0df903258951566ecc358f6696969"}, + {file = "orjson-3.10.5-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1ad1de7fef79736dde8c3554e75361ec351158a906d747bd901a52a5c9c8d24b"}, + {file = "orjson-3.10.5-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:2d97531cdfe9bdd76d492e69800afd97e5930cb0da6a825646667b2c6c6c0211"}, + {file = "orjson-3.10.5-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:d69858c32f09c3e1ce44b617b3ebba1aba030e777000ebdf72b0d8e365d0b2b3"}, + {file = "orjson-3.10.5-cp312-none-win32.whl", hash = "sha256:64c9cc089f127e5875901ac05e5c25aa13cfa5dbbbd9602bda51e5c611d6e3e2"}, + {file = "orjson-3.10.5-cp312-none-win_amd64.whl", hash = "sha256:b2efbd67feff8c1f7728937c0d7f6ca8c25ec81373dc8db4ef394c1d93d13dc5"}, + {file = "orjson-3.10.5-cp38-cp38-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:03b565c3b93f5d6e001db48b747d31ea3819b89abf041ee10ac6988886d18e01"}, + {file = "orjson-3.10.5-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:584c902ec19ab7928fd5add1783c909094cc53f31ac7acfada817b0847975f26"}, + {file = "orjson-3.10.5-cp38-cp38-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:5a35455cc0b0b3a1eaf67224035f5388591ec72b9b6136d66b49a553ce9eb1e6"}, + {file = "orjson-3.10.5-cp38-cp38-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:1670fe88b116c2745a3a30b0f099b699a02bb3482c2591514baf5433819e4f4d"}, + {file = "orjson-3.10.5-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:185c394ef45b18b9a7d8e8f333606e2e8194a50c6e3c664215aae8cf42c5385e"}, + {file = "orjson-3.10.5-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:ca0b3a94ac8d3886c9581b9f9de3ce858263865fdaa383fbc31c310b9eac07c9"}, + {file = "orjson-3.10.5-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:dfc91d4720d48e2a709e9c368d5125b4b5899dced34b5400c3837dadc7d6271b"}, + {file = "orjson-3.10.5-cp38-none-win32.whl", hash = "sha256:c05f16701ab2a4ca146d0bca950af254cb7c02f3c01fca8efbbad82d23b3d9d4"}, + {file = "orjson-3.10.5-cp38-none-win_amd64.whl", hash = "sha256:8a11d459338f96a9aa7f232ba95679fc0c7cedbd1b990d736467894210205c09"}, + {file = "orjson-3.10.5-cp39-cp39-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:85c89131d7b3218db1b24c4abecea92fd6c7f9fab87441cfc342d3acc725d807"}, + {file = "orjson-3.10.5-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fb66215277a230c456f9038d5e2d84778141643207f85336ef8d2a9da26bd7ca"}, + {file = "orjson-3.10.5-cp39-cp39-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:51bbcdea96cdefa4a9b4461e690c75ad4e33796530d182bdd5c38980202c134a"}, + {file = "orjson-3.10.5-cp39-cp39-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dbead71dbe65f959b7bd8cf91e0e11d5338033eba34c114f69078d59827ee139"}, + {file = "orjson-3.10.5-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5df58d206e78c40da118a8c14fc189207fffdcb1f21b3b4c9c0c18e839b5a214"}, + {file = "orjson-3.10.5-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:c4057c3b511bb8aef605616bd3f1f002a697c7e4da6adf095ca5b84c0fd43595"}, + {file = "orjson-3.10.5-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:b39e006b00c57125ab974362e740c14a0c6a66ff695bff44615dcf4a70ce2b86"}, + {file = "orjson-3.10.5-cp39-none-win32.whl", hash = "sha256:eded5138cc565a9d618e111c6d5c2547bbdd951114eb822f7f6309e04db0fb47"}, + {file = "orjson-3.10.5-cp39-none-win_amd64.whl", hash = "sha256:cc28e90a7cae7fcba2493953cff61da5a52950e78dc2dacfe931a317ee3d8de7"}, + {file = "orjson-3.10.5.tar.gz", hash = "sha256:7a5baef8a4284405d96c90c7c62b755e9ef1ada84c2406c24a9ebec86b89f46d"}, +] + [[package]] name = "packaging" version = "24.1" @@ -1416,7 +1491,6 @@ files = [ {file = "PyYAML-6.0.1-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:69b023b2b4daa7548bcfbd4aa3da05b3a74b772db9e23b982788168117739938"}, {file = "PyYAML-6.0.1-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:81e0b275a9ecc9c0c0c07b4b90ba548307583c125f54d5b6946cfee6360c733d"}, {file = "PyYAML-6.0.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ba336e390cd8e4d1739f42dfe9bb83a3cc2e80f567d8805e11b46f4a943f5515"}, - {file = "PyYAML-6.0.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:326c013efe8048858a6d312ddd31d56e468118ad4cdeda36c719bf5bb6192290"}, {file = "PyYAML-6.0.1-cp310-cp310-win32.whl", hash = "sha256:bd4af7373a854424dabd882decdc5579653d7868b8fb26dc7d0e99f823aa5924"}, {file = "PyYAML-6.0.1-cp310-cp310-win_amd64.whl", hash = "sha256:fd1592b3fdf65fff2ad0004b5e363300ef59ced41c2e6b3a99d4089fa8c5435d"}, {file = "PyYAML-6.0.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:6965a7bc3cf88e5a1c3bd2e0b5c22f8d677dc88a455344035f03399034eb3007"}, @@ -1424,16 +1498,8 @@ files = [ {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:42f8152b8dbc4fe7d96729ec2b99c7097d656dc1213a3229ca5383f973a5ed6d"}, {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:062582fca9fabdd2c8b54a3ef1c978d786e0f6b3a1510e0ac93ef59e0ddae2bc"}, {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d2b04aac4d386b172d5b9692e2d2da8de7bfb6c387fa4f801fbf6fb2e6ba4673"}, - {file = "PyYAML-6.0.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:e7d73685e87afe9f3b36c799222440d6cf362062f78be1013661b00c5c6f678b"}, {file = "PyYAML-6.0.1-cp311-cp311-win32.whl", hash = "sha256:1635fd110e8d85d55237ab316b5b011de701ea0f29d07611174a1b42f1444741"}, {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"}, - {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"}, - {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"}, - {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"}, - {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"}, - {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"}, - {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"}, - {file = "PyYAML-6.0.1-cp312-cp312-win_amd64.whl", hash = "sha256:0d3304d8c0adc42be59c5f8a4d9e3d7379e6955ad754aa9d6ab7a398b59dd1df"}, {file = "PyYAML-6.0.1-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:50550eb667afee136e9a77d6dc71ae76a44df8b3e51e41b77f6de2932bfe0f47"}, {file = "PyYAML-6.0.1-cp36-cp36m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:1fe35611261b29bd1de0070f0b2f47cb6ff71fa6595c077e42bd0c419fa27b98"}, {file = "PyYAML-6.0.1-cp36-cp36m-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:704219a11b772aea0d8ecd7058d0082713c3562b4e271b849ad7dc4a5c90c13c"}, @@ -1450,7 +1516,6 @@ files = [ {file = "PyYAML-6.0.1-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a0cd17c15d3bb3fa06978b4e8958dcdc6e0174ccea823003a106c7d4d7899ac5"}, {file = "PyYAML-6.0.1-cp38-cp38-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:28c119d996beec18c05208a8bd78cbe4007878c6dd15091efb73a30e90539696"}, {file = "PyYAML-6.0.1-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7e07cbde391ba96ab58e532ff4803f79c4129397514e1413a7dc761ccd755735"}, - {file = "PyYAML-6.0.1-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:49a183be227561de579b4a36efbb21b3eab9651dd81b1858589f796549873dd6"}, {file = "PyYAML-6.0.1-cp38-cp38-win32.whl", hash = "sha256:184c5108a2aca3c5b3d3bf9395d50893a7ab82a38004c8f61c258d4428e80206"}, {file = "PyYAML-6.0.1-cp38-cp38-win_amd64.whl", hash = "sha256:1e2722cc9fbb45d9b87631ac70924c11d3a401b2d7f410cc0e3bbf249f2dca62"}, {file = "PyYAML-6.0.1-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:9eb6caa9a297fc2c2fb8862bc5370d0303ddba53ba97e71f08023b6cd73d16a8"}, @@ -1458,7 +1523,6 @@ files = [ {file = "PyYAML-6.0.1-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5773183b6446b2c99bb77e77595dd486303b4faab2b086e7b17bc6bef28865f6"}, {file = "PyYAML-6.0.1-cp39-cp39-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b786eecbdf8499b9ca1d697215862083bd6d2a99965554781d0d8d1ad31e13a0"}, {file = "PyYAML-6.0.1-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bc1bf2925a1ecd43da378f4db9e4f799775d6367bdb94671027b73b393a7c42c"}, - {file = "PyYAML-6.0.1-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:04ac92ad1925b2cff1db0cfebffb6ffc43457495c9b3c39d3fcae417d7125dc5"}, {file = "PyYAML-6.0.1-cp39-cp39-win32.whl", hash = "sha256:faca3bdcf85b2fc05d06ff3fbc1f83e1391b3e724afa3feba7d13eeab355484c"}, {file = "PyYAML-6.0.1-cp39-cp39-win_amd64.whl", hash = "sha256:510c9deebc5c0225e8c96813043e62b680ba2f9c50a08d3724c7f28a747d1486"}, {file = "PyYAML-6.0.1.tar.gz", hash = "sha256:bfdf460b1736c775f2ba9f6a92bca30bc2095067b8a9d77876d1fad6cc3b4a43"}, @@ -1530,51 +1594,37 @@ python-versions = ">=3.6" files = [ {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:b42169467c42b692c19cf539c38d4602069d8c1505e97b86387fcf7afb766e1d"}, {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-macosx_13_0_arm64.whl", hash = "sha256:07238db9cbdf8fc1e9de2489a4f68474e70dffcb32232db7c08fa61ca0c7c462"}, + {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-manylinux2014_aarch64.whl", hash = "sha256:d92f81886165cb14d7b067ef37e142256f1c6a90a65cd156b063a43da1708cfd"}, {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_24_x86_64.whl", hash = "sha256:fff3573c2db359f091e1589c3d7c5fc2f86f5bdb6f24252c2d8e539d4e45f412"}, - {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-manylinux_2_24_aarch64.whl", hash = "sha256:aa2267c6a303eb483de8d02db2871afb5c5fc15618d894300b88958f729ad74f"}, - {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:840f0c7f194986a63d2c2465ca63af8ccbbc90ab1c6001b1978f05119b5e7334"}, - {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:024cfe1fc7c7f4e1aff4a81e718109e13409767e4f871443cbff3dba3578203d"}, {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-win32.whl", hash = "sha256:c69212f63169ec1cfc9bb44723bf2917cbbd8f6191a00ef3410f5a7fe300722d"}, {file = "ruamel.yaml.clib-0.2.8-cp310-cp310-win_amd64.whl", hash = "sha256:cabddb8d8ead485e255fe80429f833172b4cadf99274db39abc080e068cbcc31"}, {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:bef08cd86169d9eafb3ccb0a39edb11d8e25f3dae2b28f5c52fd997521133069"}, {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-macosx_13_0_arm64.whl", hash = "sha256:b16420e621d26fdfa949a8b4b47ade8810c56002f5389970db4ddda51dbff248"}, + {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-manylinux2014_aarch64.whl", hash = "sha256:b5edda50e5e9e15e54a6a8a0070302b00c518a9d32accc2346ad6c984aacd279"}, {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_24_x86_64.whl", hash = "sha256:25c515e350e5b739842fc3228d662413ef28f295791af5e5110b543cf0b57d9b"}, - {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-manylinux_2_24_aarch64.whl", hash = "sha256:1707814f0d9791df063f8c19bb51b0d1278b8e9a2353abbb676c2f685dee6afe"}, - {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:46d378daaac94f454b3a0e3d8d78cafd78a026b1d71443f4966c696b48a6d899"}, - {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:09b055c05697b38ecacb7ac50bdab2240bfca1a0c4872b0fd309bb07dc9aa3a9"}, {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-win32.whl", hash = "sha256:53a300ed9cea38cf5a2a9b069058137c2ca1ce658a874b79baceb8f892f915a7"}, {file = "ruamel.yaml.clib-0.2.8-cp311-cp311-win_amd64.whl", hash = "sha256:c2a72e9109ea74e511e29032f3b670835f8a59bbdc9ce692c5b4ed91ccf1eedb"}, {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:ebc06178e8821efc9692ea7544aa5644217358490145629914d8020042c24aa1"}, {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-macosx_13_0_arm64.whl", hash = "sha256:edaef1c1200c4b4cb914583150dcaa3bc30e592e907c01117c08b13a07255ec2"}, + {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-manylinux2014_aarch64.whl", hash = "sha256:7048c338b6c86627afb27faecf418768acb6331fc24cfa56c93e8c9780f815fa"}, {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d176b57452ab5b7028ac47e7b3cf644bcfdc8cacfecf7e71759f7f51a59e5c92"}, - {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-manylinux_2_24_aarch64.whl", hash = "sha256:1dc67314e7e1086c9fdf2680b7b6c2be1c0d8e3a8279f2e993ca2a7545fecf62"}, - {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:3213ece08ea033eb159ac52ae052a4899b56ecc124bb80020d9bbceeb50258e9"}, - {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:aab7fd643f71d7946f2ee58cc88c9b7bfc97debd71dcc93e03e2d174628e7e2d"}, - {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-win32.whl", hash = "sha256:5c365d91c88390c8d0a8545df0b5857172824b1c604e867161e6b3d59a827eaa"}, - {file = "ruamel.yaml.clib-0.2.8-cp312-cp312-win_amd64.whl", hash = "sha256:1758ce7d8e1a29d23de54a16ae867abd370f01b5a69e1a3ba75223eaa3ca1a1b"}, {file = "ruamel.yaml.clib-0.2.8-cp36-cp36m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:a5aa27bad2bb83670b71683aae140a1f52b0857a2deff56ad3f6c13a017a26ed"}, {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:c58ecd827313af6864893e7af0a3bb85fd529f862b6adbefe14643947cfe2942"}, {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-macosx_12_0_arm64.whl", hash = "sha256:f481f16baec5290e45aebdc2a5168ebc6d35189ae6fea7a58787613a25f6e875"}, - {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-manylinux_2_24_aarch64.whl", hash = "sha256:77159f5d5b5c14f7c34073862a6b7d34944075d9f93e681638f6d753606c6ce6"}, + {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-manylinux2014_aarch64.whl", hash = "sha256:3fcc54cb0c8b811ff66082de1680b4b14cf8a81dce0d4fbf665c2265a81e07a1"}, {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:7f67a1ee819dc4562d444bbafb135832b0b909f81cc90f7aa00260968c9ca1b3"}, - {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:4ecbf9c3e19f9562c7fdd462e8d18dd902a47ca046a2e64dba80699f0b6c09b7"}, - {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:87ea5ff66d8064301a154b3933ae406b0863402a799b16e4a1d24d9fbbcbe0d3"}, {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-win32.whl", hash = "sha256:75e1ed13e1f9de23c5607fe6bd1aeaae21e523b32d83bb33918245361e9cc51b"}, {file = "ruamel.yaml.clib-0.2.8-cp37-cp37m-win_amd64.whl", hash = "sha256:3f215c5daf6a9d7bbed4a0a4f760f3113b10e82ff4c5c44bec20a68c8014f675"}, {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:1b617618914cb00bf5c34d4357c37aa15183fa229b24767259657746c9077615"}, {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-macosx_12_0_arm64.whl", hash = "sha256:a6a9ffd280b71ad062eae53ac1659ad86a17f59a0fdc7699fd9be40525153337"}, - {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-manylinux_2_24_aarch64.whl", hash = "sha256:305889baa4043a09e5b76f8e2a51d4ffba44259f6b4c72dec8ca56207d9c6fe1"}, + {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-manylinux2014_aarch64.whl", hash = "sha256:665f58bfd29b167039f714c6998178d27ccd83984084c286110ef26b230f259f"}, {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:700e4ebb569e59e16a976857c8798aee258dceac7c7d6b50cab63e080058df91"}, - {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:e2b4c44b60eadec492926a7270abb100ef9f72798e18743939bdbf037aab8c28"}, - {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:e79e5db08739731b0ce4850bed599235d601701d5694c36570a99a0c5ca41a9d"}, {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-win32.whl", hash = "sha256:955eae71ac26c1ab35924203fda6220f84dce57d6d7884f189743e2abe3a9fbe"}, {file = "ruamel.yaml.clib-0.2.8-cp38-cp38-win_amd64.whl", hash = "sha256:56f4252222c067b4ce51ae12cbac231bce32aee1d33fbfc9d17e5b8d6966c312"}, {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:03d1162b6d1df1caa3a4bd27aa51ce17c9afc2046c31b0ad60a0a96ec22f8001"}, {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-macosx_12_0_arm64.whl", hash = "sha256:bba64af9fa9cebe325a62fa398760f5c7206b215201b0ec825005f1b18b9bccf"}, - {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-manylinux_2_24_aarch64.whl", hash = "sha256:a1a45e0bb052edf6a1d3a93baef85319733a888363938e1fc9924cb00c8df24c"}, + {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-manylinux2014_aarch64.whl", hash = "sha256:9eb5dee2772b0f704ca2e45b1713e4e5198c18f515b52743576d196348f374d3"}, {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-manylinux_2_5_x86_64.manylinux1_x86_64.whl", hash = "sha256:da09ad1c359a728e112d60116f626cc9f29730ff3e0e7db72b9a2dbc2e4beed5"}, - {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:184565012b60405d93838167f425713180b949e9d8dd0bbc7b49f074407c5a8b"}, - {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:a75879bacf2c987c003368cf14bed0ffe99e8e85acfa6c0bfffc21a090f16880"}, {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-win32.whl", hash = "sha256:84b554931e932c46f94ab306913ad7e11bba988104c5cff26d90d03f68258cd5"}, {file = "ruamel.yaml.clib-0.2.8-cp39-cp39-win_amd64.whl", hash = "sha256:25ac8c08322002b06fa1d49d1646181f0b2c72f5cbc15a85e80b4c30a544bb15"}, {file = "ruamel.yaml.clib-0.2.8.tar.gz", hash = "sha256:beb2e0404003de9a4cab9753a8805a8fe9320ee6673136ed7f04255fe60bb512"}, @@ -1960,4 +2010,4 @@ cli = ["typer", "zeroconf"] [metadata] lock-version = "2.0" python-versions = "^3.11" -content-hash = "d59fd92728002871b73473a21d8289695b198a78e3f00683f537c321a559732f" +content-hash = "9733649b2d2f967b932ef953a54b775cdafdd26a789127decde25418b552bf8d" diff --git a/pyproject.toml b/pyproject.toml index 36af14c3..1546c3e4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,9 +29,11 @@ aiohttp = ">=3.0.0" awesomeversion = ">=22.1.0" backoff = ">=2.2.0" cachetools = ">=4.0.0" +mashumaro = "^3.13" +orjson = ">=3.9.8" python = "^3.11" -yarl = ">=1.6.0" typer = {version = "^0.12.3", optional = true, extras = ["all"]} +yarl = ">=1.6.0" zeroconf = {version = "^0.132.2", optional = true, extras = ["all"]} [tool.poetry.extras] @@ -66,7 +68,7 @@ plugins = ["covdefaults"] source = ["wled"] [tool.coverage.report] -fail_under = 53 +fail_under = 25 show_missing = true omit = ["src/wled/cli/*"] @@ -138,6 +140,9 @@ max-line-length = 88 [tool.pylint.DESIGN] max-attributes = 20 +[tool.pylint.TYPECHECK] +ignored-modules = ["orjson"] + [tool.pytest.ini_options] addopts = "--cov" asyncio_mode = "auto" @@ -145,6 +150,7 @@ asyncio_mode = "auto" [tool.ruff.lint] ignore = [ "ANN101", # Self... explanatory + "ANN102", # cls... clear screen? "ANN401", # Opinioated warning on disallowing dynamically typed expressions "D203", # Conflicts with other rules "D213", # Conflicts with other rules @@ -164,6 +170,9 @@ mark-parentheses = false [tool.ruff.lint.isort] known-first-party = ["wled"] +[tool.ruff.lint.flake8-type-checking] +runtime-evaluated-base-classes = ["mashumaro.mixins.orjson.DataClassORJSONMixin"] + [tool.ruff.lint.mccabe] max-complexity = 25 diff --git a/src/wled/__init__.py b/src/wled/__init__.py index 1ca172bb..aae5c93b 100644 --- a/src/wled/__init__.py +++ b/src/wled/__init__.py @@ -1,10 +1,12 @@ """Asynchronous Python client for WLED.""" +from .const import LightCapability, LiveDataOverride, NightlightMode, SyncGroup from .exceptions import ( WLEDConnectionClosedError, WLEDConnectionError, WLEDConnectionTimeoutError, WLEDError, + WLEDUnsupportedVersionError, WLEDUpgradeError, ) from .models import ( @@ -12,18 +14,17 @@ Effect, Info, Leds, - LightCapability, - Live, Nightlight, Palette, Playlist, PlaylistEntry, Preset, + Releases, Segment, State, - Sync, + UDPSync, ) -from .wled import WLED +from .wled import WLED, WLEDReleases __all__ = [ "Device", @@ -31,19 +32,24 @@ "Info", "Leds", "LightCapability", - "Live", + "LiveDataOverride", "Nightlight", + "NightlightMode", "Palette", "Playlist", "PlaylistEntry", "Preset", + "Releases", "Segment", "State", - "Sync", + "SyncGroup", + "UDPSync", "WLED", "WLEDConnectionClosedError", "WLEDConnectionError", "WLEDConnectionTimeoutError", "WLEDError", + "WLEDReleases", + "WLEDUnsupportedVersionError", "WLEDUpgradeError", ] diff --git a/src/wled/cli/__init__.py b/src/wled/cli/__init__.py index 7e0c98b4..f26ad4e7 100644 --- a/src/wled/cli/__init__.py +++ b/src/wled/cli/__init__.py @@ -1,21 +1,289 @@ """Asynchronous Python client for WLED.""" import asyncio +import sys +from typing import Annotated +import typer from rich.console import Console from rich.live import Live +from rich.panel import Panel from rich.table import Table from zeroconf import ServiceStateChange, Zeroconf from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf +from wled import WLED, WLEDReleases +from wled.exceptions import WLEDConnectionError, WLEDUnsupportedVersionError + from .async_typer import AsyncTyper cli = AsyncTyper(help="WLED CLI", no_args_is_help=True, add_completion=False) console = Console() +@cli.error_handler(WLEDConnectionError) +def connection_error_handler(_: WLEDConnectionError) -> None: + """Handle connection errors.""" + message = """ + Could not connect to the specified WLED device. Please make sure that + the device is powered on, connected to the network and that you have + specified the correct IP address or hostname. + + If you are not sure what the IP address or hostname of your WLED device + is, you can use the scan command to find it: + + wled scan + """ + panel = Panel( + message, + expand=False, + title="Connection error", + border_style="red bold", + ) + console.print(panel) + sys.exit(1) + + +@cli.error_handler(WLEDUnsupportedVersionError) +def unsupported_version_error_handler( + _: WLEDUnsupportedVersionError, +) -> None: + """Handle unsupported version errors.""" + message = """ + The specified WLED device is running an unsupported version. + + Currently only 0.14.0 and higher is supported + """ + panel = Panel( + message, + expand=False, + title="Unsupported version", + border_style="red bold", + ) + console.print(panel) + sys.exit(1) + + +@cli.command("info") +async def command_info( + host: Annotated[ + str, + typer.Option( + help="WLED device IP address or hostname", + prompt="Host address", + show_default=False, + ), + ], +) -> None: + """Show the information about the WLED device.""" + with console.status( + "[cyan]Fetching WLED device information...", spinner="toggle12" + ): + async with WLED(host) as led: + device = await led.update() + + info_table = Table(title="\nWLED device information", show_header=False) + info_table.add_column("Property", style="cyan bold") + info_table.add_column("Value", style="green") + + info_table.add_row("Name", device.info.name) + info_table.add_row("Brand", device.info.brand) + info_table.add_row("Product", device.info.product) + + info_table.add_section() + info_table.add_row("IP address", device.info.ip) + info_table.add_row("MAC address", device.info.mac_address) + if device.info.wifi: + info_table.add_row("Wi-Fi BSSID", device.info.wifi.bssid) + info_table.add_row("Wi-Fi channel", str(device.info.wifi.channel)) + info_table.add_row("Wi-Fi RSSI", f"{device.info.wifi.rssi} dBm") + info_table.add_row("Wi-Fi signal strength", f"{device.info.wifi.signal}%") + + info_table.add_section() + info_table.add_row("Version", device.info.version) + info_table.add_row("Build", str(device.info.build)) + info_table.add_row("Architecture", device.info.architecture) + info_table.add_row("Arduino version", device.info.arduino_core_version) + + info_table.add_section() + info_table.add_row("Uptime", f"{int(device.info.uptime.total_seconds())} seconds") + info_table.add_row("Free heap", f"{device.info.free_heap} bytes") + info_table.add_row("Total storage", f"{device.info.filesystem.total} bytes") + info_table.add_row("Used storage", f"{device.info.filesystem.used} bytes") + info_table.add_row("% Used storage", f"{device.info.filesystem.used_percentage}%") + + info_table.add_section() + info_table.add_row("Effect count", f"{device.info.effect_count} effects") + info_table.add_row("Palette count", f"{device.info.palette_count} palettes") + + info_table.add_section() + info_table.add_row("Sync UDP port", str(device.info.udp_port)) + info_table.add_row( + "WebSocket", + "Disabled" + if device.info.websocket is None + else f"{device.info.websocket} client(s)", + ) + + info_table.add_section() + info_table.add_row("Live", "Yes" if device.info.live else "No") + info_table.add_row("Live IP", device.info.live_ip) + info_table.add_row("Live mode", device.info.live_mode) + + info_table.add_section() + info_table.add_row("LED count", f"{device.info.leds.count} LEDs") + info_table.add_row("LED power", f"{device.info.leds.power} mA") + info_table.add_row("LED max power", f"{device.info.leds.max_power} mA") + + console.print(info_table) + + +@cli.command("effects") +async def command_effects( + host: Annotated[ + str, + typer.Option( + help="WLED device IP address or hostname", + prompt="Host address", + show_default=False, + ), + ], +) -> None: + """Show the effects on the device.""" + with console.status( + "[cyan]Fetching WLED device information...", spinner="toggle12" + ): + async with WLED(host) as led: + device = await led.update() + + table = Table(title="\nEffects on this WLED device", show_header=False) + table.add_column("Effects", style="cyan bold") + for effect in device.effects.values(): + table.add_row(effect.name) + + console.print(table) + + +@cli.command("palettes") +async def command_palettes( + host: Annotated[ + str, + typer.Option( + help="WLED device IP address or hostname", + prompt="Host address", + show_default=False, + ), + ], +) -> None: + """Show the palettes on the device.""" + with console.status( + "[cyan]Fetching WLED device information...", spinner="toggle12" + ): + async with WLED(host) as led: + device = await led.update() + + table = Table(title="\nPalettes on this WLED device", show_header=False) + table.add_column("Palette", style="cyan bold") + for palettes in device.palettes.values(): + table.add_row(palettes.name) + + console.print(table) + + +@cli.command("playlists") +async def command_playlists( + host: Annotated[ + str, + typer.Option( + help="WLED device IP address or hostname", + prompt="Host address", + show_default=False, + ), + ], +) -> None: + """Show the playlists on the device.""" + with console.status( + "[cyan]Fetching WLED device information...", spinner="toggle12" + ): + async with WLED(host) as led: + device = await led.update() + + if not device.playlists: + console.print("🚫[red] This device has no playlists") + return + + table = Table(title="\nPlaylists stored in the WLED device", show_header=False) + table.add_column("Playlist", style="cyan bold") + for playlist in device.playlists.values(): + table.add_row(playlist.name) + + console.print(table) + + +@cli.command("presets") +async def command_presets( + host: Annotated[ + str, + typer.Option( + help="WLED device IP address or hostname", + prompt="Host address", + show_default=False, + ), + ], +) -> None: + """Show the presets on the device.""" + with console.status( + "[cyan]Fetching WLED device information...", spinner="toggle12" + ): + async with WLED(host) as led: + device = await led.update() + + if not device.presets: + console.print("🚫[red] This device has no presets") + return + + table = Table(title="\nPresets stored in the WLED device") + table.add_column("Preset", style="cyan bold") + table.add_column("Quick label", style="cyan bold") + table.add_column("Active", style="green") + for preset in device.presets.values(): + table.add_row(preset.name, preset.quick_label, "Yes" if preset.on else "No") + + console.print(table) + + +@cli.command("releases") +async def command_releases() -> None: + """Show the latest release information of WLED.""" + with console.status( + "[cyan]Fetching latest release information...", spinner="toggle12" + ): + async with WLEDReleases() as releases: + latest = await releases.releases() + + table = Table( + title="\n\nFound WLED Releases", header_style="cyan bold", show_lines=True + ) + table.add_column("Release channel") + table.add_column("Latest version") + table.add_column("Release notes") + + table.add_row( + "Stable", + latest.stable, + f"https://github.com/Aircoookie/WLED/releases/v{latest.stable}", + ) + table.add_row( + "Beta", + latest.beta, + f"https://github.com/Aircoookie/WLED/releases/v{latest.beta}", + ) + + console.print(table) + + @cli.command("scan") -async def test() -> None: +async def command_scan() -> None: """Scan for WLED devices on the network.""" zeroconf = AsyncZeroconf() background_tasks = set() diff --git a/src/wled/const.py b/src/wled/const.py new file mode 100644 index 00000000..3bd57ba8 --- /dev/null +++ b/src/wled/const.py @@ -0,0 +1,56 @@ +"""Asynchronous Python client for WLED.""" + +from enum import IntEnum, IntFlag + +from awesomeversion import AwesomeVersion + +MIN_REQUIRED_VERSION = AwesomeVersion("0.14.0") + + +class LightCapability(IntFlag): + """Enumeration representing the capabilities of a light in WLED.""" + + NONE = 0 + RGB_COLOR = 1 + WHITE_CHANNEL = 2 + COLOR_TEMPERATURE = 4 + MANUAL_WHITE = 8 + + # These are not used, but are reserved for future use. + # WLED specifications documents we should expect them, + # therefore, we include them here. + RESERVED_2 = 16 + RESERVED_3 = 32 + RESERVED_4 = 64 + RESERVED_5 = 128 + + +class LiveDataOverride(IntEnum): + """Enumeration representing live override mode from WLED.""" + + OFF = 0 + ON = 1 + OFF_UNTIL_REBOOT = 2 + + +class NightlightMode(IntEnum): + """Enumeration representing nightlight mode from WLED.""" + + INSTANT = 0 + FADE = 1 + COLOR_FADE = 2 + SUNRISE = 3 + + +class SyncGroup(IntFlag): + """Bitfield for udp sync groups 1-8.""" + + NONE = 0 + GROUP1 = 1 + GROUP2 = 2 + GROUP3 = 4 + GROUP4 = 8 + GROUP5 = 16 + GROUP6 = 32 + GROUP7 = 64 + GROUP8 = 128 diff --git a/src/wled/exceptions.py b/src/wled/exceptions.py index 451ca1fd..c6386955 100644 --- a/src/wled/exceptions.py +++ b/src/wled/exceptions.py @@ -21,5 +21,9 @@ class WLEDConnectionClosedError(WLEDConnectionError): """WLED WebSocket connection has been closed.""" +class WLEDUnsupportedVersionError(WLEDError): + """WLED version is unsupported.""" + + class WLEDUpgradeError(WLEDError): """WLED upgrade exception.""" diff --git a/src/wled/models.py b/src/wled/models.py index dd612c82..3c3da212 100644 --- a/src/wled/models.py +++ b/src/wled/models.py @@ -2,101 +2,170 @@ from __future__ import annotations -from dataclasses import dataclass -from enum import IntEnum, IntFlag -from functools import lru_cache -from operator import attrgetter +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from functools import cached_property from typing import Any from awesomeversion import AwesomeVersion +from mashumaro import field_options +from mashumaro.config import BaseConfig +from mashumaro.mixins.orjson import DataClassORJSONMixin +from mashumaro.types import SerializableType, SerializationStrategy + +from .const import ( + MIN_REQUIRED_VERSION, + LightCapability, + LiveDataOverride, + NightlightMode, + SyncGroup, +) +from .exceptions import WLEDUnsupportedVersionError +from .utils import get_awesome_version + + +class AwesomeVersionSerializationStrategy(SerializationStrategy, use_annotations=True): + """Serialization strategy for AwesomeVersion objects.""" + + def serialize(self, value: AwesomeVersion | None) -> str: + """Serialize AwesomeVersion object to string.""" + if value is None: + return "" + return str(value) + + def deserialize(self, value: str) -> AwesomeVersion | None: + """Deserialize string to AwesomeVersion object.""" + version = get_awesome_version(value) + if not version.valid: + return None + return version + + +class TimedeltaSerializationStrategy(SerializationStrategy, use_annotations=True): + """Serialization strategy for timedelta objects.""" + + def serialize(self, value: timedelta) -> int: + """Serialize timedelta object to seconds.""" + return int(value.total_seconds()) -from .exceptions import WLEDError + def deserialize(self, value: int) -> timedelta: + """Deserialize integer to timedelta object.""" + return timedelta(seconds=value) -NAME_GETTER = attrgetter("name") +class TimestampSerializationStrategy(SerializationStrategy, use_annotations=True): + """Serialization strategy for datetime objects.""" -@lru_cache -def get_awesome_version(version: str) -> AwesomeVersion: - """Return a cached AwesomeVersion object.""" - return AwesomeVersion(version) + def serialize(self, value: datetime) -> float: + """Serialize datetime object to timestamp.""" + return value.timestamp() + + def deserialize(self, value: float) -> datetime: + """Deserialize timestamp to datetime object.""" + return datetime.fromtimestamp(value, tz=UTC) @dataclass -class Nightlight: - """Object holding nightlight state in WLED.""" +class Color(SerializableType): + """Object holding color information in WLED.""" + + primary: tuple[int, int, int, int] | tuple[int, int, int] + secondary: tuple[int, int, int, int] | tuple[int, int, int] | None = None + tertiary: tuple[int, int, int, int] | tuple[int, int, int] | None = None + + def _serialize(self) -> list[tuple[int, int, int, int] | tuple[int, int, int]]: + colors = [self.primary] + if self.secondary is not None: + colors.append(self.secondary) + if self.tertiary is not None: + colors.append(self.tertiary) + return colors + + @classmethod + def _deserialize( + cls, value: list[tuple[int, int, int, int] | tuple[int, int, int] | str] + ) -> Color: + # Some values in the list can be strings, which indicates that the + # color is a hex color value. + return cls( + *[ # type: ignore[arg-type] + tuple(int(color[i : i + 2], 16) for i in (1, 3, 5)) + if isinstance(color, str) + else color + for color in value + ] + ) - duration: int - fade: bool - on: bool - mode: NightlightMode - target_brightness: int - @staticmethod - def from_dict(data: dict[str, Any]) -> Nightlight: - """Return Nightlight object from WLED API response. +class BaseModel(DataClassORJSONMixin): + """Base model for all WLED models.""" - Args: - ---- - data: The data from the WLED device API. + # pylint: disable-next=too-few-public-methods + class Config(BaseConfig): + """Mashumaro configuration.""" - Returns: - ------- - A Nightlight object. + omit_none = True + serialization_strategy = { # noqa: RUF012 + AwesomeVersion: AwesomeVersionSerializationStrategy(), + datetime: TimestampSerializationStrategy(), + timedelta: TimedeltaSerializationStrategy(), + } + serialize_by_alias = True - """ - nightlight = data.get("nl", {}) - - # Handle deprecated fade property for Nightlight - mode = nightlight.get("mode") - fade = nightlight.get("fade", False) - if mode is not None: - fade = mode != NightlightMode.INSTANT - if mode is None: - mode = NightlightMode.FADE if fade else NightlightMode.INSTANT - - return Nightlight( - duration=nightlight.get("dur", 1), - fade=fade, - mode=NightlightMode(mode), - on=nightlight.get("on", False), - target_brightness=nightlight.get("tbri", 0), - ) +@dataclass(kw_only=True) +class Nightlight(BaseModel): + """Object holding nightlight state in WLED.""" -@dataclass -class Sync: - """Object holding sync state in WLED.""" + duration: int = field(default=1, metadata=field_options(alias="dur")) + """Duration of nightlight in minutes.""" - receive: bool - send: bool + mode: NightlightMode = field(default=NightlightMode.INSTANT) + """Nightlight mode (available since 0.10.2).""" - @staticmethod - def from_dict(data: dict[str, Any]) -> Sync: - """Return Sync object from WLED API response. + on: bool = field(default=False) + """Nightlight currently active.""" - Args: - ---- - data: The data from the WLED device API. + target_brightness: int = field(default=0, metadata=field_options(alias="tbri")) + """Target brightness of nightlight feature.""" - Returns: - ------- - A sync object. - """ - sync = data.get("udpn", {}) - return Sync(send=sync.get("send", False), receive=sync.get("recv", False)) +@dataclass(kw_only=True) +class UDPSync(BaseModel): + """Object holding UDP sync state in WLED. + + Missing at this point, is the `nn` field. This field allows to skip + sending a broadcast packet for the current API request; However, this field + is only used for requests and not part of the state responses. + """ + receive: bool = field(default=False, metadata=field_options(alias="recv")) + """Receive broadcast packets.""" -@dataclass -class Effect: + receive_groups: SyncGroup = field( + default=SyncGroup.NONE, metadata=field_options(alias="rgrp") + ) + """Groups to receive WLED broadcast packets from.""" + + send: bool = field(default=False, metadata=field_options(alias="send")) + """Send WLED broadcast (UDP sync) packet on state change.""" + + send_groups: SyncGroup = field( + default=SyncGroup.NONE, metadata=field_options(alias="sgrp") + ) + """Groups to send WLED broadcast packets to.""" + + +@dataclass(frozen=True, kw_only=True) +class Effect(BaseModel): """Object holding an effect in WLED.""" effect_id: int name: str -@dataclass -class Palette: +@dataclass(frozen=True, kw_only=True) +class Palette(BaseModel): """Object holding an palette in WLED. Args: @@ -113,8 +182,8 @@ class Palette: palette_id: int -@dataclass -class Segment: +@dataclass(kw_only=True) +class Segment(BaseModel): """Object holding segment state in WLED. Args: @@ -127,142 +196,133 @@ class Segment: """ - brightness: int - clones: int - color_primary: tuple[int, int, int, int] | tuple[int, int, int] - color_secondary: tuple[int, int, int, int] | tuple[int, int, int] - color_tertiary: tuple[int, int, int, int] | tuple[int, int, int] - effect: Effect - intensity: int - length: int - on: bool - palette: Palette - reverse: bool - segment_id: int - selected: bool - speed: int - start: int - stop: int - - @staticmethod - # pylint: disable-next=too-many-arguments - def from_dict( # noqa: PLR0913 - segment_id: int, - data: dict[str, Any], - *, - effects: dict[int, Effect], - palettes: dict[int, Palette], - state_on: bool, - state_brightness: int, - ) -> Segment: - """Return Segment object from WLED API response. + brightness: int = field(default=0, metadata=field_options(alias="bri")) + """Brightness of the segment.""" - Args: - ---- - segment_id: The ID of the LED strip segment. - data: The segment data received from the WLED device. - effects: An indexed dict of Effect objects. - palettes: An indexed dict of Palette objects. - state_on: Boolean the represents the on/off state of this segment. - state_brightness: The brightness level of this segment. + clones: int = field(default=-1, metadata=field_options(alias="cln")) + """The segment this segment clones.""" - Returns: - ------- - An Segment object. + color: Color | None = field(default=None, metadata=field_options(alias="col")) + """The primary, secondary (background) and tertiary colors of the segment. - """ - start = data.get("start", 0) - stop = data.get("stop", 0) - length = data.get("len", (stop - start)) - - colors = data.get("col", []) - primary_color, secondary_color, tertiary_color = (0, 0, 0) - try: - primary_color = tuple(colors.pop(0)) # type: ignore[assignment] - secondary_color = tuple(colors.pop(0)) # type: ignore[assignment] - tertiary_color = tuple(colors.pop(0)) # type: ignore[assignment] - except IndexError: - pass - - effect = effects.get(data.get("fx", 0)) or Effect(effect_id=0, name="Unknown") - palette = palettes.get(data.get("pal", 0)) or Palette( - palette_id=0, name="Unknown" - ) + Each color is an tuple of 3 or 4 bytes, which represents a RGB(W) color, + i.e. (255,170,0) or (64,64,64,64). - return Segment( - brightness=data.get("bri", state_brightness), - clones=data.get("cln", -1), - color_primary=primary_color, # type: ignore[arg-type] - color_secondary=secondary_color, # type: ignore[arg-type] - color_tertiary=tertiary_color, # type: ignore[arg-type] - effect=effect, - intensity=data.get("ix", 0), - length=length, - on=data.get("on", state_on), - palette=palette, - reverse=data.get("rev", False), - segment_id=segment_id, - selected=data.get("sel", False), - speed=data.get("sx", 0), - start=start, - stop=stop, - ) + WLED can also return hex color values as strings, this library will + automatically convert those to RGB values to keep the data consistent. + """ + effect_id: int | str = field(default=0, metadata=field_options(alias="fx")) + """ID of the effect. -@dataclass + ~ to increment, ~- to decrement, or "r" for random. + """ + + intensity: int | str = field(default=0, metadata=field_options(alias="ix")) + """Intensity of the segment. + + Effect intensity. ~ to increment, ~- to decrement. ~10 to increment by 10, + ~-10 to decrement by 10. + """ + + length: int = field(default=0, metadata=field_options(alias="len")) + """Length of the segment (stop - start). + + Stop has preference, so if it is included, length is ignored. + """ + + on: bool | None = field(default=None) + """The on/off state of the segment.""" + + palette_id: int | str = field(default=0, metadata=field_options(alias="pal")) + """ID of the palette. + + ~ to increment, ~- to decrement, or r for random. + """ + + reverse: bool = field(default=False, metadata=field_options(alias="rev")) + """ + Flips the segment (in horizontal dimension for 2D set-up), + causing animations to change direction. + """ + + segment_id: int | None = field(default=None, metadata=field_options(alias="id")) + """The ID of the segment.""" + + selected: bool = field(default=False, metadata=field_options(alias="sel")) + """ + Indicates if the segment is selected. + + Selected segments will have their state (color/FX) updated by APIs that + don't support segments (e.g. UDP sync, HTTP API). If no segment is selected, + the first segment (id:0) will behave as if selected. + + WLED will report the state of the first (lowest id) segment that is selected + to APIs (HTTP, MQTT, Blynk...), or mainseg in case no segment is selected + and for the UDP API. + + Live data is always applied to all LEDs regardless of segment configuration. + """ + + speed: int = field(default=0, metadata=field_options(alias="sx")) + """Relative effect speed. + + ~ to increment, ~- to decrement. ~10 to increment by 10, ~-10 to decrement by 10. + """ + + start: int = 0 + """LED the segment starts at. + + For 2D set-up it determines column where segment starts, + from top-left corner of the matrix. + """ + + stop: int = 0 + """LED the segment stops at, not included in range. + + If stop is set to a lower or equal value than start (setting to 0 is + recommended), the segment is invalidated and deleted. + + For 2D set-up it determines column where segment stops, + from top-left corner of the matrix. + """ + + +@dataclass(frozen=True, kw_only=True) class Leds: """Object holding leds info from WLED.""" - cct: bool - count: int - fps: int | None - light_capabilities: LightCapability | None - max_power: int - max_segments: int - power: int - rgbw: bool - wv: bool - segment_light_capabilities: list[LightCapability] | None - - @staticmethod - def from_dict(data: dict[str, Any]) -> Leds: - """Return Leds object from WLED API response. + count: int = 0 + """Total LED count.""" - Args: - ---- - data: The data from the WLED device API. + fps: int = 0 + """Current frames per second.""" - Returns: - ------- - A Leds object. + light_capabilities: LightCapability = field( + default=LightCapability.NONE, metadata=field_options(alias="lc") + ) + """Capabilities of the light.""" - """ - leds = data.get("leds", {}) - - light_capabilities = None - segment_light_capabilities = None - if "lc" in leds and "seglc" in leds: - light_capabilities = LightCapability(leds["lc"]) - segment_light_capabilities = [ - LightCapability(item) for item in leds["seglc"] - ] + max_power: int = field(default=0, metadata=field_options(alias="maxpwr")) + """Maximum power budget in milliamps for the ABL. 0 if ABL is disabled.""" - return Leds( - cct=bool(leds.get("cct")), - count=leds.get("count", 0), - fps=leds.get("fps", None), - light_capabilities=light_capabilities, - max_power=leds.get("maxpwr", 0), - max_segments=leds.get("maxseg", 0), - power=leds.get("pwr", 0), - rgbw=leds.get("rgbw", False), - segment_light_capabilities=segment_light_capabilities, - wv=bool(leds.get("wv", True)), - ) + max_segments: int = field(default=0, metadata=field_options(alias="maxseg")) + """Maximum number of segments supported by this version.""" + power: int = field(default=0, metadata=field_options(alias="pwr")) + """ + Current LED power usage in milliamps as determined by the ABL. + 0 if ABL is disabled. + """ -@dataclass -class Wifi: + segment_light_capabilities: list[LightCapability] = field( + default_factory=list, metadata=field_options(alias="seglc") + ) + """Capabilities of each segment.""" + + +@dataclass(frozen=True, kw_only=True) +class Wifi(BaseModel): """Object holding Wi-Fi information from WLED. Args: @@ -275,37 +335,14 @@ class Wifi: """ - bssid: str - channel: int - rssi: int - signal: int - - @staticmethod - def from_dict(data: dict[str, Any]) -> Wifi | None: - """Return Wifi object form WLED API response. + bssid: str = "00:00:00:00:00:00" + channel: int = 0 + rssi: int = 0 + signal: int = 0 - Args: - ---- - data: The response from the WLED API. - - Returns: - ------- - An Wifi object. - - """ - if "wifi" not in data: - return None - wifi = data.get("wifi", {}) - return Wifi( - bssid=wifi.get("bssid", "00:00:00:00:00:00"), - channel=wifi.get("channel", 0), - rssi=wifi.get("rssi", 0), - signal=wifi.get("signal", 0), - ) - -@dataclass -class Filesystem: +@dataclass(frozen=True, kw_only=True) +class Filesystem(BaseModel): """Object holding Filesystem information from WLED. Args: @@ -318,402 +355,404 @@ class Filesystem: """ - total: int - used: int - free: int - percentage: int + last_modified: datetime | None = field( + default=None, metadata=field_options(alias="pmt") + ) + """ + Last modification of the presets.json file. Not accurate after boot or + after using /edit. + """ - @staticmethod - def from_dict(data: dict[str, Any]) -> Filesystem | None: - """Return Filesystem object form WLED API response. + total: int = field(default=1, metadata=field_options(alias="t")) + """Total space of the filesystem in kilobytes.""" - Args: - ---- - data: The response from the WLED API. + used: int = field(default=1, metadata=field_options(alias="u")) + """Used space of the filesystem in kilobytes.""" - Returns: + @cached_property + def free(self) -> int: + """Return the free space of the filesystem in kilobytes. + + Returns ------- - An Filesystem object. + The free space of the filesystem. """ - if "fs" not in data: - return None - filesystem = data.get("fs", {}) - total = filesystem.get("t", 1) - used = filesystem.get("u", 1) - return Filesystem( - total=total, - used=used, - free=(total - used), - percentage=round((used / total) * 100), - ) + return self.total - self.used + @cached_property + def free_percentage(self) -> int: + """Return the free percentage of the filesystem. -@dataclass -class Info: # pylint: disable=too-many-instance-attributes - """Object holding information from WLED.""" + Returns + ------- + The free percentage of the filesystem. - architecture: str - arduino_core_version: str - brand: str - build_type: str - effect_count: int - filesystem: Filesystem | None - free_heap: int - ip: str # pylint: disable=invalid-name - leds: Leds - live_ip: str - live_mode: str - live: bool - mac_address: str - name: str - pallet_count: int - product: str - udp_port: int - uptime: int - version_id: str - version: AwesomeVersion | None - version_latest_beta: AwesomeVersion | None - version_latest_stable: AwesomeVersion | None - websocket: int | None - wifi: Wifi | None - - @staticmethod - def from_dict(data: dict[str, Any]) -> Info: - """Return Info object from WLED API response. + """ + return round((self.free / self.total) * 100) - Args: - ---- - data: The data from the WLED device API. + @cached_property + def used_percentage(self) -> int: + """Return the used percentage of the filesystem. - Returns: + Returns ------- - A info object. + The used percentage of the filesystem. """ - if (websocket := data.get("ws")) == -1: - websocket = None - - if version := data.get("ver"): - version = get_awesome_version(version) - if not version.valid: - version = None - - if version_latest_stable := data.get("version_latest_stable"): - version_latest_stable = get_awesome_version(version_latest_stable) - - if version_latest_beta := data.get("version_latest_beta"): - version_latest_beta = get_awesome_version(version_latest_beta) - - arch = data.get("arch", "Unknown") - if ( - (filesystem := Filesystem.from_dict(data)) is not None - and arch == "esp8266" - and filesystem.total - ): - if filesystem.total <= 256: - arch = "esp01" - elif filesystem.total <= 512: - arch = "esp02" - - return Info( - architecture=arch, - arduino_core_version=data.get("core", "Unknown").replace("_", "."), - brand=data.get("brand", "WLED"), - build_type=data.get("btype", "Unknown"), - effect_count=data.get("fxcount", 0), - filesystem=filesystem, - free_heap=data.get("freeheap", 0), - ip=data.get("ip", "Unknown"), - leds=Leds.from_dict(data), - live_ip=data.get("lip", "Unknown"), - live_mode=data.get("lm", "Unknown"), - live=data.get("live", False), - mac_address=data.get("mac", ""), - name=data.get("name", "WLED Light"), - pallet_count=data.get("palcount", 0), - product=data.get("product", "DIY Light"), - udp_port=data.get("udpport", 0), - uptime=data.get("uptime", 0), - version_id=data.get("vid", "Unknown"), - version=version, - version_latest_beta=version_latest_beta, - version_latest_stable=version_latest_stable, - websocket=websocket, - wifi=Wifi.from_dict(data), - ) + return round((self.used / self.total) * 100) -@dataclass -class State: +@dataclass(kw_only=True) +class Info(BaseModel): # pylint: disable=too-many-instance-attributes + """Object holding information from WLED.""" + + architecture: str = field(default="Unknown", metadata=field_options(alias="arch")) + """Name of the platform.""" + + arduino_core_version: str = field( + default="Unknown", metadata=field_options(alias="core") + ) + """Version of the underlying (Arduino core) SDK.""" + + brand: str = "WLED" + """The producer/vendor of the light. Always WLED for standard installations.""" + + build: str = field(default="Unknown", metadata=field_options(alias="vid")) + """Build ID (YYMMDDB, B = daily build index).""" + + effect_count: int = field(default=0, metadata=field_options(alias="fxcount")) + """Number of effects included.""" + + filesystem: Filesystem = field(metadata=field_options(alias="fs")) + """Info about the embedded LittleFS filesystem.""" + + free_heap: int = field(default=0, metadata=field_options(alias="freeheap")) + """Bytes of heap memory (RAM) currently available. Problematic if <10k.""" + + ip: str = "" # pylint: disable=invalid-name + """The IP address of this instance. Empty string if not connected.""" + + leds: Leds = field(default=Leds()) + """Contains info about the LED setup.""" + + live_ip: str = field(default="Unknown", metadata=field_options(alias="lip")) + """Realtime data source IP address.""" + + live_mode: str = field(default="Unknown", metadata=field_options(alias="lm")) + """Info about the realtime data source.""" + + live: bool = False + """Realtime data source active via UDP or E1.31.""" + + mac_address: str = field(default="", metadata=field_options(alias="mac")) + """ + The hexadecimal hardware MAC address of the light, + lowercase and without colons. + """ + + name: str = "WLED Light" + """Friendly name of the light. Intended for display in lists and titles.""" + + palette_count: int = field(default=0, metadata=field_options(alias="palcount")) + """Number of palettes configured.""" + + product: str = "DIY Light" + """The product name. Always FOSS for standard installations.""" + + udp_port: int = field(default=0, metadata=field_options(alias="udpport")) + """The UDP port for realtime packets and WLED broadcast.""" + + uptime: timedelta = timedelta(0) + """Uptime of the device.""" + + version: AwesomeVersion | None = field( + default=None, metadata=field_options(alias="ver") + ) + """Version of the WLED software.""" + + websocket: int | None = field(default=None, metadata=field_options(alias="ws")) + """ + Number of currently connected WebSockets clients. + `None` indicates that WebSockets are unsupported in this build. + """ + + wifi: Wifi | None = None + """Info about the Wi-Fi connection.""" + + @classmethod + def __post_deserialize__(cls, obj: Info) -> Info: + """Post deserialize hook for Info object.""" + # If the websocket is disabled in this build, the value will be -1. + # We want to represent this as None. + if obj.websocket == -1: + obj.websocket = None + + # We can tweak the architecture name based on the filesystem size. + if obj.filesystem is not None and obj.architecture == "esp8266": + if obj.filesystem.total <= 256: + obj.architecture = "esp01" + elif obj.filesystem.total <= 512: + obj.architecture = "esp02" + + return obj + + +@dataclass(kw_only=True) +class State(BaseModel): """Object holding the state of WLED.""" - brightness: int - nightlight: Nightlight - on: bool - playlist: Playlist | int | None - preset: Preset | int | None - segments: list[Segment] - sync: Sync - transition: int - lor: Live + brightness: int = field(default=1, metadata=field_options(alias="bri")) + """Brightness of the light. - @property - def playlist_active(self) -> bool: - """Return if a playlist is currently active. + If on is false, contains last brightness when light was on (aka brightness + when on is set to true). Setting bri to 0 is supported but it is + recommended to use the range 1-255 and use on: false to turn off. - Returns - ------- - True if there is currently a playlist active, False otherwise. + The state response will never have the value 0 for bri. + """ - """ - return self.playlist == -1 + nightlight: Nightlight = field(metadata=field_options(alias="nl")) + """Nightlight state.""" - @property - def preset_active(self) -> bool: - """Return if a preset is currently active. + on: bool = False + """The on/off state of the light.""" - Returns - ------- - True is a preset is currently active, False otherwise. + playlist_id: int | None = field(default=-1, metadata=field_options(alias="pl")) + """ID of currently set playlist..""" - """ - return self.preset == -1 - - @staticmethod - def from_dict( - data: dict[str, Any], - effects: dict[int, Effect], - palettes: dict[int, Palette], - presets: dict[int, Preset], - playlists: dict[int, Playlist], - ) -> State: - """Return State object from WLED API response. + preset_id: int | None = field(default=-1, metadata=field_options(alias="ps")) + """ID of currently set preset.""" - Args: - ---- - data: The state response received from the WLED device API. - effects: A dict index of effect objects. - palettes: A dict index of palette objects. - presets: A dict index of preset objects. - playlists: A dict index of playlist objects. + segments: dict[int, Segment] = field( + default_factory=dict, metadata=field_options(alias="seg") + ) + """Segments are individual parts of the LED strip.""" - Returns: - ------- - A State object. + sync: UDPSync = field(metadata=field_options(alias="udpn")) + """UDP sync state.""" - """ - brightness = data.get("bri", 1) - on = data.get("on", False) - lor = data.get("lor", 0) - - segments = [ - Segment.from_dict( - segment_id=segment_id, - data=segment, - effects=effects, - palettes=palettes, - state_on=on, - state_brightness=brightness, - ) - for segment_id, segment in enumerate(data.get("seg", [])) - ] + transition: int = 0 + """Duration of the crossfade between different colors/brightness levels. - playlist = data.get("pl", -1) - preset = data.get("ps", -1) - if presets: - playlist = playlists.get(playlist) - preset = presets.get(preset) - - return State( - brightness=brightness, - nightlight=Nightlight.from_dict(data), - on=on, - playlist=playlist, - preset=preset, - segments=segments, - sync=Sync.from_dict(data), - transition=data.get("transition", 0), - lor=Live(lor), - ) + One unit is 100ms, so a value of 4 results in atransition of 400ms. + """ + live_data_override: LiveDataOverride = field(metadata=field_options(alias="lor")) + """Live data override. -@dataclass -class Preset: + 0 is off, 1 is override until live data ends, 2 is override until ESP reboot. + """ + + @classmethod + def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]: + """Pre deserialize hook for State object.""" + # Segments are not indexes, which is suboptimal for the user. + # We will add the segment ID to the segment data and convert + # the segments list to an indexed dict. + d["seg"] = { + segment_id: segment | {"id": segment_id} + for segment_id, segment in enumerate(d.get("seg", [])) + } + return d + + @classmethod + def __post_deserialize__(cls, obj: State) -> State: + """Post deserialize hook for State object.""" + # If no playlist is active, the value will be -1. We want to represent + # this as None. + if obj.playlist_id == -1: + obj.playlist_id = None + + # If no preset is active, the value will be -1. We want to represent + # this as None. + if obj.preset_id == -1: + obj.preset_id = None + + return obj + + +@dataclass(kw_only=True) +class Preset(BaseModel): """Object representing a WLED preset.""" preset_id: int - name: str - quick_label: str | None + """The ID of the preset.""" - on: bool - transition: int - main_segment: Segment | None - segments: list[Segment] - - @staticmethod - def from_dict( - preset_id: int, - data: dict[str, Any], - effects: dict[int, Effect], - palettes: dict[int, Palette], - ) -> Preset: - """Return Preset object from WLED API response. + name: str = field(default="", metadata=field_options(alias="n")) + """The name of the preset.""" - Args: - ---- - preset_id: The ID of the preset. - data: The data from the WLED device API. - effects: A indexed dict of effect objects. - palettes: A indexed dict of palette object. + quick_label: str | None = field(default=None, metadata=field_options(alias="ql")) + """The quick label of the preset.""" - Returns: - ------- - A Preset object. + on: bool = False + """The on/off state of the preset.""" - """ - segment_data = data.get("seg", []) - if not isinstance(segment_data, list): - # Some older versions of WLED have an single segment - # instead of a list. - segment_data = [segment_data] - - segments = [ - Segment.from_dict( - segment_id=segment_id, - data=segment, - effects=effects, - palettes=palettes, - state_on=False, - state_brightness=0, - ) - for segment_id, segment in enumerate(segment_data) - ] + transition: int = 0 + """Duration of the crossfade between different colors/brightness levels. - try: - main_segment = segments[data.get("mainseg", 0)] - except IndexError: - main_segment = None - - return Preset( - main_segment=main_segment, - name=data.get("n", str(preset_id)), - on=data.get("on", False), - preset_id=preset_id, - quick_label=data.get("ql"), - segments=segments, - transition=data.get("transition", 0), - ) + One unit is 100ms, so a value of 4 results in atransition of 400ms. + """ + main_segment_id: int = field(default=0, metadata=field_options(alias="mainseg")) + """The main segment of the preset.""" -@dataclass -class PlaylistEntry: + segments: list[Segment] = field( + default_factory=list, metadata=field_options(alias="seg") + ) + """Segments are individual parts of the LED strip.""" + + @classmethod + def __post_deserialize__(cls, obj: Preset) -> Preset: + """Post deserialize hook for Preset object.""" + # If name is empty, we will replace it with the playlist ID. + if not obj.name: + obj.name = str(obj.preset_id) + return obj + + +@dataclass(frozen=True, kw_only=True) +class PlaylistEntry(BaseModel): """Object representing a entry in a WLED playlist.""" - duration: int + duration: int = field(metadata=field_options(alias="dur")) entry_id: int - preset: Preset | None + preset: int = field(metadata=field_options(alias="ps")) transition: int -@dataclass -class Playlist: +@dataclass(kw_only=True) +class Playlist(BaseModel): """Object representing a WLED playlist.""" - end: Preset | None - entries: list[PlaylistEntry] - name: str - playlist_id: int - repeat: int - shuffle: bool + end_preset_id: int | None = field(default=None, metadata=field_options(alias="end")) + """Single preset ID to apply after the playlist finished. - @staticmethod - def from_dict( - playlist_id: int, - data: dict[str, Any], - presets: dict[int, Preset], - ) -> Playlist: - """Return Playlist object from WLED API response. + Has no effect when an indefinite cycle is set. If not provided, + the light will stay on the last preset of the playlist. + """ - Args: - ---- - playlist_id: The ID of the playlist. - data: The data from the WLED device API. - presets: A list of preset objects. + entries: list[PlaylistEntry] + """List of entries in the playlist.""" - Returns: - ------- - A Playlist object. + name: str = field(default="", metadata=field_options(alias="n")) + """The name of the playlist.""" - """ - playlist = data.get("playlist", {}) - entries_durations = playlist.get("dur", []) - entries_presets = playlist.get("ps", []) - entries_transitions = playlist.get("transition", []) - - entries = [ - PlaylistEntry( - entry_id=entry_id, - duration=entries_durations[entry_id], - transition=entries_transitions[entry_id], - preset=presets.get(preset_id), + playlist_id: int + """The ID of the playlist.""" + + repeat: int = 0 + """Number of times the playlist should repeat.""" + + shuffle: bool = field(default=False, metadata=field_options(alias="r")) + """Shuffle the playlist entries.""" + + @classmethod + def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]: + """Pre deserialize hook for State object.""" + d |= d["playlist"] + # Duration, presets and transitions values are separate lists stored + # in the playlist data. We will combine those into a list of + # dictionaries, which will make it easier to work with the data. + item_count = len(d.get("ps", [])) + + # If the duration is a single value, we will convert it to a list. + # with the same length as the presets list. + if not isinstance(d["dur"], list): + d["dur"] = [d["dur"]] * item_count + + # If the transition value doesn't exists, we will set it to 0. + if "transitions" not in d: + d["transitions"] = [0] * item_count + # If the transition is a single value, we will convert it to a list. + # with the same length as the presets list. + elif not isinstance(d["transitions"], list): + d["transitions"] = [d["transitions"]] * item_count + + # Now we can easily combine the data into a list of dictionaries. + d["entries"] = [ + { + "entry_id": entry_id, + "ps": ps, + "dur": dur, + "transition": transition, + } + for entry_id, (ps, dur, transition) in enumerate( + zip(d["ps"], d["dur"], d["transitions"]) ) - for entry_id, preset_id in enumerate(entries_presets) ] - end = presets.get(playlist.get("end")) + return d - return Playlist( - playlist_id=playlist_id, - shuffle=playlist.get("r", False), - name=data.get("n", str(playlist_id)), - repeat=playlist.get("repeat", 0), - end=end, - entries=entries, - ) + @classmethod + def __post_deserialize__(cls, obj: Playlist) -> Playlist: + """Post deserialize hook for Playlist object.""" + # If name is empty, we will replace it with the playlist ID. + if not obj.name: + obj.name = str(obj.playlist_id) + return obj -class Device: +@dataclass(kw_only=True) +class Device(BaseModel): """Object holding all information of WLED.""" - effects: list[Effect] info: Info - palettes: list[Palette] - playlists: list[Playlist] - presets: list[Preset] state: State - def __init__(self, data: dict[str, Any]) -> None: - """Initialize an empty WLED device class. + effects: dict[int, Effect] = field(default_factory=dict) + palettes: dict[int, Palette] = field(default_factory=dict) + playlists: dict[int, Playlist] = field(default_factory=dict) + presets: dict[int, Preset] = field(default_factory=dict) + + @classmethod + def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]: + """Pre deserialize hook for Device object.""" + if (version := d.get("info", {}).get("ver")) and version < MIN_REQUIRED_VERSION: + msg = ( + f"Unsupported firmware version {version}. " + f"Minimum required version is {MIN_REQUIRED_VERSION}. " + f"Please update your WLED device." + ) + raise WLEDUnsupportedVersionError(msg) - Args: - ---- - data: The full API response from a WLED device. + if _effects := d.get("effects"): + d["effects"] = { + effect_id: {"effect_id": effect_id, "name": name} + for effect_id, name in enumerate(_effects) + } - Raises: - ------ - WLEDError: In case the given API response is incomplete in a way - that a Device object cannot be constructed from it. + if _palettes := d.get("palettes"): + d["palettes"] = { + palette_id: {"palette_id": palette_id, "name": name} + for palette_id, name in enumerate(_palettes) + } - """ - self._indexed_effects: dict[int, Effect] = {} - self._indexed_palettes: dict[int, Palette] = {} - self._indexed_presets: dict[int, Preset] = {} - self._indexed_playlists: dict[int, Playlist] = {} - - self.effects = [] - self.palettes = [] - self.playlists = [] - self.presets = [] - - # Check if all elements are in the passed dict, else raise an Error - if any( - k not in data and data[k] is not None - for k in ("effects", "palettes", "info", "state") - ): - msg = "WLED data is incomplete, cannot construct device object" - raise WLEDError(msg) - self.update_from_dict(data) + if _presets := d.get("presets"): + _presets = _presets.copy() + # The preset data contains both presets and playlists, + # we split those out, so we can handle those correctly. + d["presets"] = { + int(preset_id): preset | {"preset_id": int(preset_id)} + for preset_id, preset in _presets.items() + if "playlist" not in preset + or "ps" not in preset["playlist"] + or not preset["playlist"]["ps"] + } + # Nobody cares about 0. + d["presets"].pop(0, None) + + d["playlists"] = { + int(playlist_id): playlist | {"playlist_id": int(playlist_id)} + for playlist_id, playlist in _presets.items() + if "playlist" in playlist + and "ps" in playlist["playlist"] + and playlist["playlist"]["ps"] + } + # Nobody cares about 0. + d["playlists"].pop(0, None) + + return d def update_from_dict(self, data: dict[str, Any]) -> Device: """Return Device object from WLED API response. @@ -729,40 +768,35 @@ def update_from_dict(self, data: dict[str, Any]) -> Device: """ if _effects := data.get("effects"): - self._indexed_effects = { - effect_id: Effect(effect_id=effect_id, name=effect) - for effect_id, effect in enumerate(_effects) + self.effects = { + effect_id: Effect(effect_id=effect_id, name=name) + for effect_id, name in enumerate(_effects) } - self.effects = sorted(self._indexed_effects.values(), key=NAME_GETTER) if _palettes := data.get("palettes"): - self._indexed_palettes = { - palette_id: Palette(palette_id=palette_id, name=palette) - for palette_id, palette in enumerate(_palettes) + self.palettes = { + palette_id: Palette(palette_id=palette_id, name=name) + for palette_id, name in enumerate(_palettes) } - self.palettes = sorted(self._indexed_palettes.values(), key=NAME_GETTER) if _presets := data.get("presets"): # The preset data contains both presets and playlists, # we split those out, so we can handle those correctly. - self._indexed_presets = { + self.presets = { int(preset_id): Preset.from_dict( - int(preset_id), - preset, - self._indexed_effects, - self._indexed_palettes, + preset | {"preset_id": int(preset_id)}, ) for preset_id, preset in _presets.items() if "playlist" not in preset - or not ("ps" in preset["playlist"] and preset["playlist"]["ps"]) + or "ps" not in preset["playlist"] + or not preset["playlist"]["ps"] } # Nobody cares about 0. - self._indexed_presets.pop(0, None) - self.presets = sorted(self._indexed_presets.values(), key=NAME_GETTER) + self.presets.pop(0, None) - self._indexed_playlists = { + self.playlists = { int(playlist_id): Playlist.from_dict( - int(playlist_id), playlist, self._indexed_presets + playlist | {"playlist_id": int(playlist_id)} ) for playlist_id, playlist in _presets.items() if "playlist" in playlist @@ -770,54 +804,20 @@ def update_from_dict(self, data: dict[str, Any]) -> Device: and playlist["playlist"]["ps"] } # Nobody cares about 0. - self._indexed_playlists.pop(0, None) - self.playlists = sorted(self._indexed_playlists.values(), key=NAME_GETTER) + self.playlists.pop(0, None) if _info := data.get("info"): self.info = Info.from_dict(_info) if _state := data.get("state"): - self.state = State.from_dict( - _state, - self._indexed_effects, - self._indexed_palettes, - self._indexed_presets, - self._indexed_playlists, - ) + self.state = State.from_dict(_state) return self -class LightCapability(IntFlag): - """Enumeration representing the capabilities of a light in WLED.""" - - NONE = 0 - RGB_COLOR = 1 - WHITE_CHANNEL = 2 - COLOR_TEMPERATURE = 4 - MANUAL_WHITE = 8 - - # These are not used, but are reserved for future use. - # WLED specifications documents we should expect them, - # therefore, we include them here. - RESERVED_2 = 16 - RESERVED_3 = 32 - RESERVED_4 = 64 - RESERVED_5 = 128 - - -class Live(IntEnum): - """Enumeration representing live override mode from WLED.""" - - OFF = 0 - ON = 1 - OFF_UNTIL_REBOOT = 2 - - -class NightlightMode(IntEnum): - """Enumeration representing nightlight mode from WLED.""" +@dataclass(frozen=True, kw_only=True) +class Releases(BaseModel): + """Object holding WLED releases information.""" - INSTANT = 0 - FADE = 1 - COLOR_FADE = 2 - SUNRISE = 3 + beta: AwesomeVersion | None + stable: AwesomeVersion | None diff --git a/src/wled/utils.py b/src/wled/utils.py new file mode 100644 index 00000000..bdfd3dff --- /dev/null +++ b/src/wled/utils.py @@ -0,0 +1,11 @@ +"""Asynchronous Python client for WLED.""" + +from functools import lru_cache + +from awesomeversion import AwesomeVersion + + +@lru_cache +def get_awesome_version(version: str) -> AwesomeVersion: + """Return a cached AwesomeVersion object.""" + return AwesomeVersion(version) diff --git a/src/wled/wled.py b/src/wled/wled.py index d689a515..17c53584 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -3,16 +3,13 @@ from __future__ import annotations import asyncio -import json import socket -from contextlib import suppress from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Self import aiohttp import backoff -from awesomeversion import AwesomeVersion, AwesomeVersionException -from cachetools import TTLCache +import orjson from yarl import URL from .exceptions import ( @@ -23,13 +20,14 @@ WLEDError, WLEDUpgradeError, ) -from .models import Device, Live, Playlist, Preset +from .models import Device, Playlist, Preset, Releases if TYPE_CHECKING: from collections.abc import Callable, Sequence + from awesomeversion import AwesomeVersion -VERSION_CACHE: TTLCache[str, str | None] = TTLCache(maxsize=16, ttl=7200) + from .const import LiveDataOverride @dataclass @@ -43,8 +41,6 @@ class WLED: _client: aiohttp.ClientWebSocketResponse | None = None _close_session: bool = False _device: Device | None = None - _supports_si_request: bool | None = None - _supports_presets: bool | None = None @property def connected(self) -> bool: @@ -204,24 +200,16 @@ async def request( if content_type == "application/json": raise WLEDError( response.status, - json.loads(contents.decode("utf8")), + orjson.loads(contents), ) raise WLEDError( response.status, {"message": contents.decode("utf8")}, ) + response_data = await response.text() if "application/json" in content_type: - response_data = await response.json() - if ( - method == "POST" - and uri == "/json/state" - and self._device is not None - and data is not None - ): - self._device.update_from_dict(data={"state": response_data}) - else: - response_data = await response.text() + response_data = orjson.loads(response_data) except asyncio.TimeoutError as exception: msg = f"Timeout occurred while connecting to WLED device at {self.host}" @@ -230,6 +218,14 @@ async def request( msg = f"Error occurred while communicating with WLED device at {self.host}" raise WLEDConnectionError(msg) from exception + if "application/json" in content_type and ( + method == "POST" + and uri == "/json/state" + and self._device is not None + and data is not None + ): + self._device.update_from_dict(data={"state": response_data}) + return response_data @backoff.on_exception( @@ -238,107 +234,40 @@ async def request( max_tries=3, logger=None, ) - async def update(self, *, full_update: bool = False) -> Device: + async def update(self) -> Device: """Get all information about the device in a single call. This method updates all WLED information available with a single API call. - Args: - ---- - full_update: Force a full update from the WLED Device. - - Returns: + Returns ------- WLED Device data. - Raises: + Raises ------ WLEDEmptyResponseError: The WLED device returned an empty response. """ - if self._device is None or full_update: - if not (data := await self.request("/json")): - msg = ( - f"WLED device at {self.host} returned an empty API" - " response on full update", - ) - raise WLEDEmptyResponseError(msg) - - # Try to get presets, introduced in WLED 0.11 - try: - presets = await self.request("/presets.json") - data["presets"] = presets - self._supports_presets = True - except WLEDError: - self._supports_presets = False - - with suppress(WLEDError): - versions = await self.get_wled_versions_from_github() - data["info"].update(versions) - - self._device = Device(data) - - # Try to figure out if this version supports - # a single info and state call - try: - self._supports_si_request = self._device.info.version >= AwesomeVersion( - "0.10.0", - ) - except AwesomeVersionException: - # Could be a manual build one? Lets poll for it - try: - await self.request("/json/si") - self._supports_si_request = True - except WLEDError: - self._supports_si_request = False - - return self._device - - if self._supports_presets: - if not (presets := await self.request("/presets.json")): - msg = ( - f"WLED device at {self.host} returned an empty API" - " response on presets update", - ) - raise WLEDEmptyResponseError(msg) - self._device.update_from_dict({"presets": presets}) - - # Handle legacy state and update in separate requests - if not self._supports_si_request: - if not (info := await self.request("/json/info")): - msg = ( - f"WLED device at {self.host} returned an empty API" - " response on info update", - ) - raise WLEDEmptyResponseError(msg) - - if not (state := await self.request("/json/state")): - msg = ( - f"WLED device {self.host} returned an empty API" - " response on state update", - ) - raise WLEDEmptyResponseError(msg) - - with suppress(WLEDError): - versions = await self.get_wled_versions_from_github() - info.update(versions) - - self._device.update_from_dict({"info": info, "state": state}) - return self._device - - if not (state_info := await self.request("/json/si")): + if not (data := await self.request("/json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on state & info update", + " response on full update", ) raise WLEDEmptyResponseError(msg) - with suppress(WLEDError): - versions = await self.get_wled_versions_from_github() - state_info["info"].update(versions) + if not (presets := await self.request("/presets.json")): + msg = ( + f"WLED device at {self.host} returned an empty API" + " response on presets update", + ) + raise WLEDEmptyResponseError(msg) + data["presets"] = presets - self._device.update_from_dict(state_info) + if not self._device: + self._device = Device.from_dict(data) + else: + self._device.update_from_dict(data) return self._device @@ -439,7 +368,7 @@ async def segment( # noqa: PLR0912, PLR0913 msg = "Unable to communicate with WLED to get the current state" raise WLEDError(msg) - state = {} + state = {} # type: ignore[var-annotated] segment = { "bri": brightness, "cln": clones, @@ -456,24 +385,12 @@ async def segment( # noqa: PLR0912, PLR0913 "sx": speed, } - # > WLED 0.10.0, does not support segment control on/bri. - # Luckily, the same release introduced si requests. - # Therefore, we can use that capability check to decide. - if not self._supports_si_request: - # This device does not support on/bri in the segment - del segment["on"] - del segment["bri"] - state = { - "bri": brightness, - "on": on, - } - # Find effect if it was based on a name if effect is not None and isinstance(effect, str): segment["fx"] = next( ( item.effect_id - for item in self._device.effects + for item in self._device.effects.values() if item.name.lower() == effect.lower() ), None, @@ -484,7 +401,7 @@ async def segment( # noqa: PLR0912, PLR0913 segment["pal"] = next( ( item.palette_id - for item in self._device.palettes + for item in self._device.palettes.values() if item.name.lower() == palette.lower() ), None, @@ -499,12 +416,20 @@ async def segment( # noqa: PLR0912, PLR0913 if color_primary is not None: colors.append(color_primary) elif color_secondary is not None or color_tertiary is not None: - colors.append(self._device.state.segments[segment_id].color_primary) + if clrs := self._device.state.segments[segment_id].color: + colors.append(clrs.primary) + else: + colors.append((0, 0, 0)) if color_secondary is not None: colors.append(color_secondary) elif color_tertiary is not None: - colors.append(self._device.state.segments[segment_id].color_secondary) + if ( + clrs := self._device.state.segments[segment_id].color + ) and clrs.secondary: + colors.append(clrs.secondary) + else: + colors.append((0, 0, 0)) if color_tertiary is not None: colors.append(color_tertiary) @@ -514,7 +439,7 @@ async def segment( # noqa: PLR0912, PLR0913 if segment: segment["id"] = segment_id - state["seg"] = [segment] # type: ignore[assignment] + state["seg"] = [segment] if transition is not None: state["tt"] = transition @@ -550,7 +475,7 @@ async def preset(self, preset: int | str | Preset) -> None: preset = next( ( item.preset_id - for item in self._device.presets + for item in self._device.presets.values() if item.name.lower() == preset.lower() ), preset, @@ -574,7 +499,7 @@ async def playlist(self, playlist: int | str | Playlist) -> None: playlist = next( ( item.playlist_id - for item in self._device.playlists + for item in self._device.playlists.values() if item.name.lower() == playlist.lower() ), playlist, @@ -585,7 +510,7 @@ async def playlist(self, playlist: int | str | Playlist) -> None: await self.request("/json/state", method="POST", data={"ps": playlist}) - async def live(self, live: Live) -> None: + async def live(self, live: LiveDataOverride) -> None: """Set the live override mode on a WLED device. Args: @@ -742,8 +667,49 @@ async def upgrade(self, *, version: str | AwesomeVersion) -> None: ) raise WLEDConnectionError(msg) from exception + async def reset(self) -> None: + """Reboot WLED device.""" + await self.request("/reset") + + async def close(self) -> None: + """Close open client (WebSocket) session.""" + await self.disconnect() + if self.session and self._close_session: + await self.session.close() + + async def __aenter__(self) -> Self: + """Async enter. + + Returns + ------- + The WLED object. + + """ + return self + + async def __aexit__(self, *_exc_info: object) -> None: + """Async exit. + + Args: + ---- + _exc_info: Exec type. + + """ + await self.close() + + +@dataclass +class WLEDReleases: + """Get version information for WLED.""" + + request_timeout: float = 8.0 + session: aiohttp.client.ClientSession | None = None + + _client: aiohttp.ClientWebSocketResponse | None = None + _close_session: bool = False + @backoff.on_exception(backoff.expo, WLEDConnectionError, max_tries=3, logger=None) - async def get_wled_versions_from_github(self) -> dict[str, str | None]: + async def releases(self) -> Releases: """Fetch WLED version information from GitHub. Returns @@ -760,14 +726,9 @@ async def get_wled_versions_from_github(self) -> dict[str, str | None]: version information. """ - with suppress(KeyError): - return { - "version_latest_stable": VERSION_CACHE["stable"], - "version_latest_beta": VERSION_CACHE["beta"], - } - if self.session is None: - return {"version_latest_stable": None, "version_latest_beta": None} + self.session = aiohttp.ClientSession() + self._close_session = True try: async with asyncio.timeout(self.request_timeout): @@ -776,26 +737,28 @@ async def get_wled_versions_from_github(self) -> dict[str, str | None]: headers={"Accept": "application/json"}, ) except asyncio.TimeoutError as exception: - msg = "Timeout occurred while fetching WLED version information from GitHub" + msg = ( + "Timeout occurred while fetching WLED releases information from GitHub" + ) raise WLEDConnectionTimeoutError(msg) from exception except (aiohttp.ClientError, socket.gaierror) as exception: - msg = "Timeout occurred while communicating with GitHub for WLED version" + msg = "Timeout occurred while communicating with GitHub for WLED releases" raise WLEDConnectionError(msg) from exception content_type = response.headers.get("Content-Type", "") + contents = await response.read() if response.status // 100 in [4, 5]: - contents = await response.read() response.close() if content_type == "application/json": - raise WLEDError(response.status, json.loads(contents.decode("utf8"))) + raise WLEDError(response.status, orjson.loads(contents)) raise WLEDError(response.status, {"message": contents.decode("utf8")}) if "application/json" not in content_type: - msg = "No JSON response from GitHub while retrieving version information" + msg = "No JSON response from GitHub while retrieving WLED releases" raise WLEDError(msg) - releases = await response.json() + releases = orjson.loads(contents) version_latest = None version_latest_beta = None for release in releases: @@ -812,22 +775,13 @@ async def get_wled_versions_from_github(self) -> dict[str, str | None]: if version_latest is not None and version_latest_beta is not None: break - # Cache results - VERSION_CACHE["stable"] = version_latest - VERSION_CACHE["beta"] = version_latest_beta - - return { - "version_latest_stable": version_latest, - "version_latest_beta": version_latest_beta, - } - - async def reset(self) -> None: - """Reboot WLED device.""" - await self.request("/reset") + return Releases( + beta=version_latest_beta, + stable=version_latest, + ) async def close(self) -> None: - """Close open client (WebSocket) session.""" - await self.disconnect() + """Close open client session.""" if self.session and self._close_session: await self.session.close() @@ -836,7 +790,7 @@ async def __aenter__(self) -> Self: Returns ------- - The WLED object. + The WLEDReleases object. """ return self diff --git a/tests/test_wled.py b/tests/test_wled.py index a953d9ad..12b306bc 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -1,8 +1,6 @@ """Tests for `wled.WLED`.""" import asyncio -from collections.abc import Generator -from unittest.mock import patch import aiohttp import pytest @@ -12,17 +10,6 @@ from wled.exceptions import WLEDConnectionError, WLEDError -@pytest.fixture(autouse=True) -def _mock_get_version_from_github() -> Generator[None, None, None]: - """Patch out connection to GitHub.""" - with patch( - "wled.WLED.get_wled_versions_from_github", - return_value={"version_latest_stable": None, "version_latest_beta": None}, - side_effect=WLEDConnectionError, - ): - yield - - @pytest.mark.asyncio async def test_json_request(aresponses: ResponsesMockServer) -> None: """Test JSON response is handled correctly.""" @@ -174,594 +161,3 @@ async def test_http_error500(aresponses: ResponsesMockServer) -> None: wled = WLED("example.com", session=session) with pytest.raises(WLEDError): assert await wled.request("/") - - -@pytest.mark.asyncio -async def test_state_on(aresponses: ResponsesMockServer) -> None: - """Test request of current WLED device state.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.9.1"}}' - ), - ), - ) - aresponses.add( - "example.com", - "/json/info", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"ver": "1.0"}', - ), - ) - aresponses.add( - "example.com", - "/json/state", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"on": false}', - ), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.state.on - device = await wled.update() - assert not device.state.on - - -@pytest.mark.asyncio -async def test_state_on_si_request(aresponses: ResponsesMockServer) -> None: - """Test request of current WLED device state.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - aresponses.add( - "example.com", - "/json/si", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"state": {"on": false},"info": {"ver": "1.0"}}', - ), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.state.on - device = await wled.update() - assert not device.state.on - - -@pytest.mark.asyncio -async def test_empty_responses(aresponses: ResponsesMockServer) -> None: - """Test empty responses for WLED device state.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.8.6"}}' - ), - ), - ) - aresponses.add( - "example.com", - "/json/info", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text="{}", - ), - ) - aresponses.add( - "example.com", - "/json/info", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"ver": "1.0"}', - ), - ) - aresponses.add( - "example.com", - "/json/state", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text="{}", - ), - ) - aresponses.add( - "example.com", - "/json/info", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"ver": "1.0"}', - ), - ) - aresponses.add( - "example.com", - "/json/state", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"on": false}', - ), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - await wled.update() - - -@pytest.mark.asyncio -async def test_empty_si_responses(aresponses: ResponsesMockServer) -> None: - """Test request of current WLED device state.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - aresponses.add( - "example.com", - "/json/si", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text="{}", - ), - ) - aresponses.add( - "example.com", - "/json/si", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"state": {"on": false}, "info": {"ver": "1.0"}}', - ), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - await wled.update() - - -@pytest.mark.asyncio -async def test_empty_full_responses(aresponses: ResponsesMockServer) -> None: - """Test failure handling of full data request WLED device state.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text="{}", - ), - ) - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - - -@pytest.mark.asyncio -async def test_si_request_version_based(aresponses: ResponsesMockServer) -> None: - """Test for supporting SI requests based on version data.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - assert wled._supports_si_request # pylint: disable=protected-access - - -@pytest.mark.asyncio -async def test_not_supporting_si_request_version_based( - aresponses: ResponsesMockServer, -) -> None: - """Test for supporting SI requests based on version data.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.9.1"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - assert not wled._supports_si_request # pylint: disable=protected-access - - -@pytest.mark.asyncio -async def test_si_request_probing_based(aresponses: ResponsesMockServer) -> None: - """Test for supporting SI requests based on probing.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "INVALID VERSION NUMBER"}}' - ), - ), - ) - - aresponses.add( - "example.com", - "/json/si", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text='{"yes": true}', - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - assert wled._supports_si_request # pylint: disable=protected-access - - -@pytest.mark.asyncio -async def test_not_supporting_si_request_probing_based( - aresponses: ResponsesMockServer, -) -> None: - """Test for supporting SI requests based on probing.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {"ver": "INVALID VERSION NUMBER"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - assert not wled._supports_si_request # pylint: disable=protected-access - - -@pytest.mark.asyncio -async def test_info_contains_wv_true(aresponses: ResponsesMockServer) -> None: - """Test for determining if wv is used and set to true.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {' - '"ver": "0.11.1",' - '"leds": {' - '"count": 120,' - '"rgbw": true,' - '"wv": true' - "}}}" - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.info.leds.wv - - -@pytest.mark.asyncio -async def test_info_contains_wv_false(aresponses: ResponsesMockServer) -> None: - """Test for determining if wv is used and set to false.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {' - '"ver": "0.11.1",' - '"leds": {' - '"count": 120,' - '"rgbw": true,' - '"wv": false' - "}}}" - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert not device.info.leds.wv - - -@pytest.mark.asyncio -async def test_info_contains_no_wv(aresponses: ResponsesMockServer) -> None: - """Test for determining if wv is used and set to false.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true},' - '"effects": [], "palettes": [],' - '"info": {' - '"ver": "0.8.4",' - '"leds": {' - '"count": 120,' - '"rgbw": true' - "}}}" - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.info.leds.wv - - -@pytest.mark.asyncio -async def test_live_override_state_off(aresponses: ResponsesMockServer) -> None: - """Test request of current WLED live override mode.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true, "lor": 0},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.state.lor == 0 - - -@pytest.mark.asyncio -async def test_live_override_state_on(aresponses: ResponsesMockServer) -> None: - """Test request of current WLED live override mode.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true, "lor": 1},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.state.lor == 1 - - -@pytest.mark.asyncio -async def test_live_override_state_off_until_reboot( - aresponses: ResponsesMockServer, -) -> None: - """Test request of current WLED live override mode.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=( - '{"state": {"on": true, "lor": 2},' - '"effects": [], "palettes": [],' - '"info": {"ver": "0.10.0"}}' - ), - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.state.lor == 2 - - -@pytest.mark.asyncio -async def test_reset(aresponses: ResponsesMockServer) -> None: - """Test rebooting WLED device works.""" - aresponses.add( - "example.com", - "/reset", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "text/html"}, - text="", - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.reset() - - -@pytest.mark.asyncio -async def test_detect_esp01( - aresponses: ResponsesMockServer, -) -> None: - """Test if the ESP01 is detected correctly.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=""" - { - "state": {"on": true}, - "effects": [], - "palettes": [], - "info": { - "ver": "0.10.0", - "arch": "esp8266", - "fs": { - "t": 256 - } - } - } - """, - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.info.architecture == "esp01" - - -@pytest.mark.asyncio -async def test_detect_esp02( - aresponses: ResponsesMockServer, -) -> None: - """Test if the ESP01 is detected correctly.""" - aresponses.add( - "example.com", - "/json", - "GET", - aresponses.Response( - status=200, - headers={"Content-Type": "application/json"}, - text=""" - { - "state": {"on": true}, - "effects": [], - "palettes": [], - "info": { - "ver": "0.10.0", - "arch": "esp8266", - "fs": { - "t": 512 - } - } - } - """, - ), - ) - - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert device.info.architecture == "esp02"