light9 Changeset - 44fc477970bf

Changeset - 44fc477970bf

Parent rev.

Child rev.

[Not reviewed]

default

0 13 0

drewp@bigasterisk.com - 8 months ago 2024-05-19 06:19:04
drewp@bigasterisk.com

cleanup imports, reformats

13 files changed with 71 insertions and 63 deletions:

pdm.lock

pyproject.toml

src/light9/collector/collector.py

src/light9/collector/collector_client_asyncio.py

src/light9/collector/device.py

src/light9/collector/weblisteners.py

src/light9/effect/edit.py

src/light9/effect/effect_functions.py

src/light9/effect/effecteval2.py

src/light9/effect/scale.py

src/light9/effect/sequencer/eval_faders.py

src/light9/effect/sequencer/service.py

src/light9/effect/settings.py

0 comments (0 inline, 0 general)

pdm.lock

➞

Show inline comments

 # This file is @generated by PDM.
 # It is not intended for manual editing.
 [metadata]
 groups = ["default", "dev"]
 strategy = ["cross_platform", "inherit_metadata"]
 lock_version = "4.4.1"
-content_hash = "sha256:5393d5c679935ba9f042f2b4f4d6efd58dbf03519b2e9f25e08f1e9d421e52f1"
+content_hash = "sha256:6fac24ed6ab93fd328a74d22973a01c77d9de5b4a0b22cd111a884afd99f235f"
 [[package]]
 name = "aiohttp"
 version = "3.9.5"
 requires_python = ">=3.8"
 summary = "Async http client/server framework (asyncio)"
 groups = ["default", "dev"]
 dependencies = [
     "aiosignal>=1.1.2",
     "attrs>=17.3.0",
     "frozenlist>=1.1.1",
     "multidict<7.0,>=4.5",
     "yarl<2.0,>=1.0",
+]
 files = [
     {file = "aiohttp-3.9.5-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:e0ae53e33ee7476dd3d1132f932eeb39bf6125083820049d06edcdca4381f342"},
     {file = "aiohttp-3.9.5-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:c088c4d70d21f8ca5c0b8b5403fe84a7bc8e024161febdd4ef04575ef35d474d"},
     {file = "aiohttp-3.9.5-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:639d0042b7670222f33b0028de6b4e2fad6451462ce7df2af8aee37dcac55424"},
     {file = "aiohttp-3.9.5-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f26383adb94da5e7fb388d441bf09c61e5e35f455a3217bfd790c6b6bc64b2ee"},
     {file = "aiohttp-3.9.5-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:66331d00fb28dc90aa606d9a54304af76b335ae204d1836f65797d6fe27f1ca2"},
     {file = "aiohttp-3.9.5-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:4ff550491f5492ab5ed3533e76b8567f4b37bd2995e780a1f46bca2024223233"},
     {file = "aiohttp-3.9.5-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f22eb3a6c1080d862befa0a89c380b4dafce29dc6cd56083f630073d102eb595"},
     {file = "aiohttp-3.9.5-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a81b1143d42b66ffc40a441379387076243ef7b51019204fd3ec36b9f69e77d6"},
     {file = "aiohttp-3.9.5-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:f64fd07515dad67f24b6ea4a66ae2876c01031de91c93075b8093f07c0a2d93d"},
     {file = "aiohttp-3.9.5-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:93e22add827447d2e26d67c9ac0161756007f152fdc5210277d00a85f6c92323"},
     {file = "aiohttp-3.9.5-cp311-cp311-musllinux_1_1_ppc64le.whl", hash = "sha256:55b39c8684a46e56ef8c8d24faf02de4a2b2ac60d26cee93bc595651ff545de9"},
     {file = "aiohttp-3.9.5-cp311-cp311-musllinux_1_1_s390x.whl", hash = "sha256:4715a9b778f4293b9f8ae7a0a7cef9829f02ff8d6277a39d7f40565c737d3771"},
     {file = "aiohttp-3.9.5-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:afc52b8d969eff14e069a710057d15ab9ac17cd4b6753042c407dcea0e40bf75"},
     {file = "aiohttp-3.9.5-cp311-cp311-win32.whl", hash = "sha256:b3df71da99c98534be076196791adca8819761f0bf6e08e07fd7da25127150d6"},
     {file = "aiohttp-3.9.5-cp311-cp311-win_amd64.whl", hash = "sha256:88e311d98cc0bf45b62fc46c66753a83445f5ab20038bcc1b8a1cc05666f428a"},
     {file = "aiohttp-3.9.5-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:c7a4b7a6cf5b6eb11e109a9755fd4fda7d57395f8c575e166d363b9fc3ec4678"},
     {file = "aiohttp-3.9.5-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:0a158704edf0abcac8ac371fbb54044f3270bdbc93e254a82b6c82be1ef08f3c"},
     {file = "aiohttp-3.9.5-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:d153f652a687a8e95ad367a86a61e8d53d528b0530ef382ec5aaf533140ed00f"},
     {file = "aiohttp-3.9.5-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:82a6a97d9771cb48ae16979c3a3a9a18b600a8505b1115cfe354dfb2054468b4"},
     {file = "aiohttp-3.9.5-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:60cdbd56f4cad9f69c35eaac0fbbdf1f77b0ff9456cebd4902f3dd1cf096464c"},
     {file = "aiohttp-3.9.5-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:8676e8fd73141ded15ea586de0b7cda1542960a7b9ad89b2b06428e97125d4fa"},
     {file = "aiohttp-3.9.5-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:da00da442a0e31f1c69d26d224e1efd3a1ca5bcbf210978a2ca7426dfcae9f58"},
     {file = "aiohttp-3.9.5-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:18f634d540dd099c262e9f887c8bbacc959847cfe5da7a0e2e1cf3f14dbf2daf"},
     {file = "aiohttp-3.9.5-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:320e8618eda64e19d11bdb3bd04ccc0a816c17eaecb7e4945d01deee2a22f95f"},
     {file = "aiohttp-3.9.5-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:2faa61a904b83142747fc6a6d7ad8fccff898c849123030f8e75d5d967fd4a81"},
     {file = "aiohttp-3.9.5-cp312-cp312-musllinux_1_1_ppc64le.whl", hash = "sha256:8c64a6dc3fe5db7b1b4d2b5cb84c4f677768bdc340611eca673afb7cf416ef5a"},
     {file = "aiohttp-3.9.5-cp312-cp312-musllinux_1_1_s390x.whl", hash = "sha256:393c7aba2b55559ef7ab791c94b44f7482a07bf7640d17b341b79081f5e5cd1a"},
     {file = "aiohttp-3.9.5-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:c671dc117c2c21a1ca10c116cfcd6e3e44da7fcde37bf83b2be485ab377b25da"},
     {file = "aiohttp-3.9.5-cp312-cp312-win32.whl", hash = "sha256:5a7ee16aab26e76add4afc45e8f8206c95d1d75540f1039b84a03c3b3800dd59"},
     {file = "aiohttp-3.9.5-cp312-cp312-win_amd64.whl", hash = "sha256:5ca51eadbd67045396bc92a4345d1790b7301c14d1848feaac1d6a6c9289e888"},
     {file = "aiohttp-3.9.5.tar.gz", hash = "sha256:edea7d15772ceeb29db4aff55e482d4bcfb6ae160ce144f2682de02f6d693551"},
+]
 [[package]]
 name = "aiosignal"
 version = "1.3.1"
 requires_python = ">=3.7"
 summary = "aiosignal: a list of registered asynchronous callbacks"
 groups = ["default", "dev"]
 dependencies = [
     "frozenlist>=1.1.0",
+]
 files = [
     {file = "aiosignal-1.3.1-py3-none-any.whl", hash = "sha256:f8376fb07dd1e86a584e4fcdec80b36b7f81aac666ebc724e2c090300dd83b17"},
     {file = "aiosignal-1.3.1.tar.gz", hash = "sha256:54cd96e15e1649b75d6c87526a6ff0b6c1b0dd3459f43d9ca11d48c339b68cfc"},
+]
 [[package]]
 name = "alsa-midi"
 version = "1.0.2"
 summary = "Python interface for ALSA MIDI sequencer"
 groups = ["default"]
 dependencies = [
     "cffi>=1.14.0",
+]
 files = [
     {file = "alsa-midi-1.0.2.tar.gz", hash = "sha256:456573ba98edde04e8dedda12166fe4066d7abca07b87e97690b394ca1e6b22b"},
     {file = "alsa_midi-1.0.2-cp311-cp311-manylinux_2_17_i686.whl", hash = "sha256:52651d916b29566feb5ee0790e3edd51baab06fcffe3036407ee48767424943d"},
     {file = "alsa_midi-1.0.2-cp311-cp311-manylinux_2_17_x86_64.whl", hash = "sha256:27e65c83a55b3a3f0748f6e9067f30e8dec8af355070d82a967dee47381b9e5a"},
     {file = "alsa_midi-1.0.2-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:89b8457bdf2d144f3c3d33442869b136360dc3569f4c355c7c97238bea7a558c"},
     {file = "alsa_midi-1.0.2-py3-none-any.whl", hash = "sha256:eb9614a9545b8567a60d919b7e86736cfc52db2e106755ff1f5cdb3ca9f461e2"},
+]
 [[package]]
 name = "anyio"
 version = "4.3.0"
 requires_python = ">=3.8"
 summary = "High level compatibility layer for multiple asynchronous event loop implementations"
 groups = ["default", "dev"]
 dependencies = [
     "idna>=2.8",
     "sniffio>=1.1",
+]
 files = [
     {file = "anyio-4.3.0-py3-none-any.whl", hash = "sha256:048e05d0f6caeed70d731f3db756d35dcc1f35747c8c403364a8332c630441b8"},
     {file = "anyio-4.3.0.tar.gz", hash = "sha256:f75253795a87df48568485fd18cdd2a3fa5c4f7c5be8e5e36637733fce06fed6"},
+]
 [[package]]
 name = "asttokens"
 version = "2.4.1"
@@ @@ -2222,132 +2222,119 @@ files = [ @@
 name = "yapf"
 version = "0.40.2"
 requires_python = ">=3.7"
 summary = "A formatter for Python code"
 groups = ["dev"]
 dependencies = [
     "importlib-metadata>=6.6.0",
     "platformdirs>=3.5.1",
     "tomli>=2.0.1",
+]
 files = [
     {file = "yapf-0.40.2-py3-none-any.whl", hash = "sha256:adc8b5dd02c0143108878c499284205adb258aad6db6634e5b869e7ee2bd548b"},
     {file = "yapf-0.40.2.tar.gz", hash = "sha256:4dab8a5ed7134e26d57c1647c7483afb3f136878b579062b786c9ba16b94637b"},
+]
 [[package]]
 name = "yappi"
 version = "1.6.0"
 requires_python = ">=3.6"
 summary = "Yet Another Python Profiler"
 groups = ["default"]
 files = [
     {file = "yappi-1.6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:e9bc33b8ec9bce8b2575a4c3878b3cd223d08eb728669924699e5ac937e7b515"},
     {file = "yappi-1.6.0-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a1cb70d46827a137350fb84b8fddecd7acec0a11834c763209875788b738f873"},
     {file = "yappi-1.6.0-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a2cefe387bc747afcf0b26c9548e242113e17fac3de2674d900e97eb58a328f6"},
     {file = "yappi-1.6.0-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:acfbf4c80b6ee0513ad35a6e4a1f633aa2f93357517f9701aed6ad8cd56544d4"},
     {file = "yappi-1.6.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:e4959c1dcfb6da8441d05915bfbb9c697e9f11655568f65b87c341e543bd65d5"},
     {file = "yappi-1.6.0-cp311-cp311-win32.whl", hash = "sha256:88dee431bba79866692f444110695133181efb2a6969ab63752f4424787f79c8"},
     {file = "yappi-1.6.0-cp311-cp311-win_amd64.whl", hash = "sha256:8ddbe1475964f145b028f8bf120a58903d8f6c7bdd1be0a16c1471ba2d8646ca"},
     {file = "yappi-1.6.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:1ba7d12c18bc0d092463ad126a95a1b2b8c261c47b0e3bd4cb2fd7479469141c"},
     {file = "yappi-1.6.0-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:fb5bbb4c6b996736554cb8f41e7fb6d5ee6096b7c4f54112cce8cf953a92c0a4"},
     {file = "yappi-1.6.0-cp312-cp312-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c12da5f310d81779056566259fef644a9c14ac1ec9a2b1b8a3fc62beb4ca6980"},
     {file = "yappi-1.6.0-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:677d992c41b239441eee399ac39ea7601010ddb5acb92bf997de7589f9ee2cc1"},
     {file = "yappi-1.6.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:d58e60aac43041d109f0931917204ef02ac01004b9579fe173f2847fbc69655b"},
     {file = "yappi-1.6.0-cp312-cp312-win32.whl", hash = "sha256:a6797f189b7b89154d6c7c53ac769a22f0adb7bd88ea5b8f6c65106a286afad6"},
     {file = "yappi-1.6.0-cp312-cp312-win_amd64.whl", hash = "sha256:cdaa263ba667aac9bf7bdc0d96fd10e2761a287f01fe87dc136f064ab7696af3"},
     {file = "yappi-1.6.0.tar.gz", hash = "sha256:a9aaf72009d8c03067294151ee0470ac7a6dfa7b33baab40b198d6c1ef00430a"},
+]
 [[package]]
 name = "yarl"
 version = "1.9.4"
 requires_python = ">=3.7"
 summary = "Yet another URL library"
 groups = ["default", "dev"]
 dependencies = [
     "idna>=2.0",
     "multidict>=4.0",
+]
 files = [
     {file = "yarl-1.9.4-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:35a2b9396879ce32754bd457d31a51ff0a9d426fd9e0e3c33394bf4b9036b099"},
     {file = "yarl-1.9.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:4c7d56b293cc071e82532f70adcbd8b61909eec973ae9d2d1f9b233f3d943f2c"},
     {file = "yarl-1.9.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:d8a1c6c0be645c745a081c192e747c5de06e944a0d21245f4cf7c05e457c36e0"},
     {file = "yarl-1.9.4-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:4b3c1ffe10069f655ea2d731808e76e0f452fc6c749bea04781daf18e6039525"},
     {file = "yarl-1.9.4-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:549d19c84c55d11687ddbd47eeb348a89df9cb30e1993f1b128f4685cd0ebbf8"},
     {file = "yarl-1.9.4-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:a7409f968456111140c1c95301cadf071bd30a81cbd7ab829169fb9e3d72eae9"},
     {file = "yarl-1.9.4-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e23a6d84d9d1738dbc6e38167776107e63307dfc8ad108e580548d1f2c587f42"},
     {file = "yarl-1.9.4-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:d8b889777de69897406c9fb0b76cdf2fd0f31267861ae7501d93003d55f54fbe"},
     {file = "yarl-1.9.4-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:03caa9507d3d3c83bca08650678e25364e1843b484f19986a527630ca376ecce"},
     {file = "yarl-1.9.4-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:4e9035df8d0880b2f1c7f5031f33f69e071dfe72ee9310cfc76f7b605958ceb9"},
     {file = "yarl-1.9.4-cp311-cp311-musllinux_1_1_ppc64le.whl", hash = "sha256:c0ec0ed476f77db9fb29bca17f0a8fcc7bc97ad4c6c1d8959c507decb22e8572"},
     {file = "yarl-1.9.4-cp311-cp311-musllinux_1_1_s390x.whl", hash = "sha256:ee04010f26d5102399bd17f8df8bc38dc7ccd7701dc77f4a68c5b8d733406958"},
     {file = "yarl-1.9.4-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:49a180c2e0743d5d6e0b4d1a9e5f633c62eca3f8a86ba5dd3c471060e352ca98"},
     {file = "yarl-1.9.4-cp311-cp311-win32.whl", hash = "sha256:81eb57278deb6098a5b62e88ad8281b2ba09f2f1147c4767522353eaa6260b31"},
     {file = "yarl-1.9.4-cp311-cp311-win_amd64.whl", hash = "sha256:d1d2532b340b692880261c15aee4dc94dd22ca5d61b9db9a8a361953d36410b1"},
     {file = "yarl-1.9.4-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:0d2454f0aef65ea81037759be5ca9947539667eecebca092733b2eb43c965a81"},
     {file = "yarl-1.9.4-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:44d8ffbb9c06e5a7f529f38f53eda23e50d1ed33c6c869e01481d3fafa6b8142"},
     {file = "yarl-1.9.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:aaaea1e536f98754a6e5c56091baa1b6ce2f2700cc4a00b0d49eca8dea471074"},
     {file = "yarl-1.9.4-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:3777ce5536d17989c91696db1d459574e9a9bd37660ea7ee4d3344579bb6f129"},
     {file = "yarl-1.9.4-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:9fc5fc1eeb029757349ad26bbc5880557389a03fa6ada41703db5e068881e5f2"},
     {file = "yarl-1.9.4-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:ea65804b5dc88dacd4a40279af0cdadcfe74b3e5b4c897aa0d81cf86927fee78"},
     {file = "yarl-1.9.4-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:aa102d6d280a5455ad6a0f9e6d769989638718e938a6a0a2ff3f4a7ff8c62cc4"},
     {file = "yarl-1.9.4-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:09efe4615ada057ba2d30df871d2f668af661e971dfeedf0c159927d48bbeff0"},
     {file = "yarl-1.9.4-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:008d3e808d03ef28542372d01057fd09168419cdc8f848efe2804f894ae03e51"},
     {file = "yarl-1.9.4-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:6f5cb257bc2ec58f437da2b37a8cd48f666db96d47b8a3115c29f316313654ff"},
     {file = "yarl-1.9.4-cp312-cp312-musllinux_1_1_ppc64le.whl", hash = "sha256:992f18e0ea248ee03b5a6e8b3b4738850ae7dbb172cc41c966462801cbf62cf7"},
     {file = "yarl-1.9.4-cp312-cp312-musllinux_1_1_s390x.whl", hash = "sha256:0e9d124c191d5b881060a9e5060627694c3bdd1fe24c5eecc8d5d7d0eb6faabc"},
     {file = "yarl-1.9.4-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:3986b6f41ad22988e53d5778f91855dc0399b043fc8946d4f2e68af22ee9ff10"},
     {file = "yarl-1.9.4-cp312-cp312-win32.whl", hash = "sha256:4b21516d181cd77ebd06ce160ef8cc2a5e9ad35fb1c5930882baff5ac865eee7"},
     {file = "yarl-1.9.4-cp312-cp312-win_amd64.whl", hash = "sha256:a9bd00dc3bc395a662900f33f74feb3e757429e545d831eef5bb280252631984"},
     {file = "yarl-1.9.4-py3-none-any.whl", hash = "sha256:928cecb0ef9d5a7946eb6ff58417ad2fe9375762382f1bf5c55e61645f2c43ad"},
     {file = "yarl-1.9.4.tar.gz", hash = "sha256:566db86717cf8080b99b58b083b773a908ae40f06681e87e589a976faf8246bf"},
+]
 [[package]]
 name = "zipp"
 version = "3.18.1"
 requires_python = ">=3.8"
 summary = "Backport of pathlib-compatible object wrapper for zip files"
 groups = ["dev"]
 files = [
     {file = "zipp-3.18.1-py3-none-any.whl", hash = "sha256:206f5a15f2af3dbaee80769fb7dc6f249695e940acca08dfb2a4769fe61e538b"},
     {file = "zipp-3.18.1.tar.gz", hash = "sha256:2884ed22e7d8961de1c9a05142eb69a247f120291bc0206a00a7642f09b5b715"},
+]
 [[package]]
 name = "zmq"
 version = "0.0.0"
 summary = "You are probably looking for pyzmq."
 groups = ["default"]
 dependencies = [
     "pyzmq",
+]
 files = [
     {file = "zmq-0.0.0.tar.gz", hash = "sha256:6b1a1de53338646e8c8405803cffb659e8eb7bb02fff4c9be62a7acfac8370c9"},
     {file = "zmq-0.0.0.zip", hash = "sha256:21cfc6be254c9bc25e4dabb8a3b2006a4227966b7b39a637426084c8dc6901f7"},
+]
 [[package]]
 name = "zope-interface"
 version = "6.3"
 requires_python = ">=3.7"
 summary = "Interfaces for Python"
 groups = ["default"]
 dependencies = [
     "setuptools",
+]
 files = [
     {file = "zope.interface-6.3-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:600101f43a7582d5b9504a7c629a1185a849ce65e60fca0f6968dfc4b76b6d39"},
     {file = "zope.interface-6.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4d6b229f5e1a6375f206455cc0a63a8e502ed190fe7eb15e94a312dc69d40299"},
     {file = "zope.interface-6.3-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:10cde8dc6b2fd6a1d0b5ca4be820063e46ddba417ab82bcf55afe2227337b130"},
     {file = "zope.interface-6.3-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:40aa8c8e964d47d713b226c5baf5f13cdf3a3169c7a2653163b17ff2e2334d10"},
     {file = "zope.interface-6.3-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d165d7774d558ea971cb867739fb334faf68fc4756a784e689e11efa3becd59e"},
     {file = "zope.interface-6.3-cp311-cp311-win_amd64.whl", hash = "sha256:69dedb790530c7ca5345899a1b4cb837cc53ba669051ea51e8c18f82f9389061"},
     {file = "zope.interface-6.3-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:8d407e0fd8015f6d5dfad481309638e1968d70e6644e0753f229154667dd6cd5"},
     {file = "zope.interface-6.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:72d5efecad16c619a97744a4f0b67ce1bcc88115aa82fcf1dc5be9bb403bcc0b"},
     {file = "zope.interface-6.3-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:567d54c06306f9c5b6826190628d66753b9f2b0422f4c02d7c6d2b97ebf0a24e"},
     {file = "zope.interface-6.3-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:483e118b1e075f1819b3c6ace082b9d7d3a6a5eb14b2b375f1b80a0868117920"},
     {file = "zope.interface-6.3-cp312-cp312-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2bb78c12c1ad3a20c0d981a043d133299117b6854f2e14893b156979ed4e1d2c"},
     {file = "zope.interface-6.3-cp312-cp312-win_amd64.whl", hash = "sha256:ad4524289d8dbd6fb5aa17aedb18f5643e7d48358f42c007a5ee51a2afc2a7c5"},
     {file = "zope.interface-6.3.tar.gz", hash = "sha256:f83d6b4b22262d9a826c3bd4b2fbfafe1d0000f085ef8e44cd1328eea274ae6a"},
+]

pyproject.toml

➞

Show inline comments

 [project]
 name = "light9"
 version = "0.0.1"
 description = ""
 authors = [{ name = "Drew Perttula", email = "drewp@bigasterisk.com" }]
 license = {text = "MIT"}
 dependencies = [
     "pillow",
     "pyjade",
     "pyserial",
     "pyusb",
     "twisted[tls]>=22.10.0",
     "txzmq",
     "aiohttp>=3.8.1",
     "coloredlogs>=15.0.1",
     "colormath>=3.0.0",
     "flask==2.2.4", # workaround for pydmxcontrol
     "ipython>=8.13.2",
     "louie>=2.0",
     "moviepy>=1.0.3",
     "noise>=1.2.2",
     "prometheus-client>=0.14.1",
     "pydmxcontrol>=2.0.0",
     "PyGObject>=3.42.1",
     "python-dateutil>=2.8.2",
     "rdflib>=6.3.2",
     "requests>=2.30.0",
     "rx>=3.2.0",
     "sse-starlette>=0.10.3",
     "starlette-exporter>=0.12.0",
     "starlette>=0.27.0",
     "statprof>=0.1.2",
     "toposort>=1.10",
     "udmx-pyusb>=2.0.0",
     "uvicorn[standard]>=0.17.6",
     "watchdog>=2.1.7",
     "webcolors>=1.11.1",
     "scipy>=1.9.3",
     "braillegraph>=0.6",
     "tenacity>=8.2.2",
     "zmq>=0.0.0",
     "mido>=1.2.10",
     "alsa-midi>=1.0.1",
     "treq>=22.2.0",
     "light9 @ file:///${PROJECT_ROOT}/",
     "python-debouncer>=0.1.4",
     "pytest>=8.2.0",
     "avro>=1.11.3",
     "fastavro>=1.9.4",
     "yappi>=1.6.0",
+]
 requires-python = ">=3.11"
 readme = "README.md"
 [project.urls]
 Homepage = "https://bigasterisk.com/light9/"
 [tool.pdm]
 [tool.pdm.dev-dependencies]
 dev = [
     "coverage>=7.2.5",
     "flake8>=6.0.0",
     "freezegun>=1.2.2",
     "hunter>=3.6.1",
     "ipdb>=0.13.13",
     "mock>=5.0.2",
     "yapf>=0.33.0",
     "pydeps>=1.12.5",
     "nose2>=0.13.0",
     "pytest-watch>=4.2.0",
     "-e file:///my/proj/rdfdb#egg=rdfdb",
+]
 [[tool.pdm.source]]
 name = "pypi"
 url = "https://pypi.org/simple"
 verify_ssl = true

src/light9/collector/collector.py

➞

Show inline comments

 import logging
 import time
 from typing import Dict, List, Set, Tuple, cast
 from light9.typedgraph import typedValue
 from prometheus_client import Summary
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import URIRef
 from light9.collector.device import resolve, toOutputAttrs
 from light9.collector.output import Output as OutputInstance
 from light9.collector.weblisteners import WebListeners
 from light9.effect.settings import DeviceSettings
 from light9.namespaces import L9, RDF
 from light9.newtypes import (ClientSessionType, ClientType, DeviceAttr, DeviceClass, DeviceSetting, DeviceUri, DmxIndex, DmxMessageIndex, OutputAttr,
                              OutputRange, OutputUri, OutputValue, UnixTime, VTUnion, uriTail)
 from light9.newtypes import (
     ClientSessionType,
     ClientType,
     DeviceAttr,
     DeviceClass,
     DeviceUri,
     DmxIndex,
     DmxMessageIndex,
     OutputAttr,
     OutputRange,
     OutputUri,
     OutputValue,
     UnixTime,
     VTUnion,
     uriTail,
+)
 from light9.typedgraph import typedValue
 log = logging.getLogger('collector')
 STAT_SETATTR = Summary('set_attr', 'setAttr calls')
 def makeDmxMessageIndex(base: DmxIndex, offset: DmxIndex) -> DmxMessageIndex:
     return DmxMessageIndex(base + offset - 1)
 def _outputMap(graph: SyncedGraph, outputs: Set[OutputUri]) -> Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]]:
     """From rdf config graph, compute a map of
        (device, outputattr) : (output, index)
     that explains which output index to set for any device update.
     """
     ret = cast(Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]], {})
     for dc in graph.subjects(RDF.type, L9['DeviceClass']):
         log.info('  mapping devices of class %s', dc)
         for dev in graph.subjects(RDF.type, dc):
             dev = cast(DeviceUri, dev)
             log.info('    💡 mapping device %s', dev)
             universe = typedValue(OutputUri, graph, dev, L9['dmxUniverse'])
             if universe not in outputs:
                 raise ValueError(f'{dev=} is configured to be in {universe=}, but we have no Output for that universe')
             try:
                 dmxBase = typedValue(DmxIndex, graph, dev, L9['dmxBase'])
             except ValueError:
                 raise ValueError('no :dmxBase for %s' % dev)
             for row in sorted(graph.objects(dc, L9['attr']), key=str):
                 outputAttr = typedValue(OutputAttr, graph, row, L9['outputAttr'])
                 offset = typedValue(DmxIndex, graph, row, L9['dmxOffset'])
                 index = makeDmxMessageIndex(dmxBase, offset)
                 ret[(dev, outputAttr)] = (universe, index)
                 log.info(f'      {uriTail(outputAttr):15} maps to {uriTail(universe)} index {index}')
     return ret
 class Collector:
     """receives setAttrs calls; combines settings; renders them into what outputs like; calls Output.update"""
     def __init__(self, graph: SyncedGraph, outputs: List[OutputInstance], listeners: WebListeners, clientTimeoutSec: float = 10):
         self.graph = graph
         self.outputs = outputs
         self.listeners = listeners
         self.clientTimeoutSec = clientTimeoutSec
         self._initTime = time.time()
         self._outputByUri: Dict[OutputUri, OutputInstance] = {}
         self._deviceType: Dict[DeviceUri, DeviceClass] = {}
         self.remapOut: Dict[Tuple[DeviceUri, OutputAttr], OutputRange] = {}
         self.graph.addHandler(self._compile)
         # rename to activeSessons ?
         self.lastRequest: Dict[Tuple[ClientType, ClientSessionType], Tuple[UnixTime, Dict[Tuple[DeviceUri, DeviceAttr], VTUnion]]] = {}
         # (dev, devAttr): value to use instead of 0
         self.stickyAttrs: Dict[Tuple[DeviceUri, DeviceAttr], VTUnion] = {}
     def _compile(self):
         log.info('Collector._compile:')
         self._outputByUri = self._compileOutputByUri()
         self._outputMap = _outputMap(self.graph, set(self._outputByUri.keys()))
         self._deviceType.clear()
         self.remapOut.clear()
         for dc in self.graph.subjects(RDF.type, L9['DeviceClass']):
             dc = cast(DeviceClass, dc)
             for dev in self.graph.subjects(RDF.type, dc):
                 dev = cast(DeviceUri, dev)
                 self._deviceType[dev] = dc
                 self._compileRemapForDevice(dev)
     def _compileOutputByUri(self) -> Dict[OutputUri, OutputInstance]:
         ret = {}
         for output in self.outputs:
             ret[OutputUri(output.uri)] = output
         return ret
     def _compileRemapForDevice(self, dev: DeviceUri):
         for remap in self.graph.objects(dev, L9['outputAttrRange']):
             attr = typedValue(OutputAttr, self.graph, remap, L9['outputAttr'])
             start = typedValue(float, self.graph, remap, L9['start'])
             end = typedValue(float, self.graph, remap, L9['end'])
             self.remapOut[(dev, attr)] = OutputRange((start, end))
     @STAT_SETATTR.time()
     def setAttrs(self, client: ClientType, clientSession: ClientSessionType, settings: DeviceSettings, sendTime: UnixTime):
         """
         Given DeviceSettings, we resolve conflicting values,
         process them into output attrs, and call Output.update
         to send the new outputs.
         client is a string naming the type of client.
         (client, clientSession) is a unique client instance.
         clientSession is deprecated.
         Each client session's last settings will be forgotten
         after clientTimeoutSec.
         """

src/light9/collector/collector_client_asyncio.py

➞

Show inline comments

 import asyncio
 import json
 import logging
 import time
 from light9 import networking
 from light9.effect.settings import DeviceSettings
 import zmq.asyncio
 from prometheus_client import Summary
 from light9.effect.settings import DeviceSettings
 log = logging.getLogger('coll_client')
 ZMQ_SEND = Summary('zmq_send', 'calls')
 def toCollectorJson(client, session, settings: DeviceSettings) -> str:
     assert isinstance(settings, DeviceSettings)
     return json.dumps({
         'settings': settings.asList(),
         'client': client,
         'clientSession': session,
         'sendTime': time.time(),
     })
 class _Sender:
     def __init__(self):
         self.context = zmq.asyncio.Context()
         self.socket = self.context.socket(zmq.PUB)
         self.socket.connect('tcp://127.0.0.1:9203')  #todo: tie to :collectorZmq in graph
         # old version used: 'tcp://%s:%s' % (service.host, service.port)
     @ZMQ_SEND.time()
     async def send(self, client: str, session: str, settings: DeviceSettings):
         msg = toCollectorJson(client, session, settings).encode('utf8')
         # log.info(f'zmq send {len(msg)}')
         await self.socket.send_multipart([b'setAttr', msg])
 _sender = _Sender()
 sendToCollector = _sender.send

src/light9/collector/device.py

➞

Show inline comments

 import logging
 from typing import Dict, List, Any, TypeVar, cast
 from light9.namespaces import L9
 from typing import Dict, List, cast
 import colormath.color_conversions
 from colormath.color_objects import CMYColor, sRGBColor
 from rdflib import Literal, URIRef
 from webcolors import hex_to_rgb, rgb_to_hex
 from colormath.color_objects import sRGBColor, CMYColor
 import colormath.color_conversions
 from light9.newtypes import VT, DeviceClass, HexColor, OutputAttr, OutputValue, DeviceUri, DeviceAttr, VTUnion
 from light9.namespaces import L9
 from light9.newtypes import (
     DeviceAttr,
     DeviceClass,
     HexColor,
     OutputAttr,
     OutputValue,
     VTUnion,
+)
 log = logging.getLogger('device')
 class Device:
     pass
 class ChauvetColorStrip(Device):
     """
      device attrs:
        color
     """
 class Mini15(Device):
     """
     plan:
       device attrs
         rx, ry
         color
         gobo
         goboShake
         imageAim (configured with a file of calibration data)
     """
 def clamp255(x):
     return min(255, max(0, x))
 def _8bit(f):
     if not isinstance(f, (int, float)):
         raise TypeError(repr(f))
     return clamp255(int(f * 255))
 def _maxColor(values: List[HexColor]) -> HexColor:
     rgbs = [hex_to_rgb(v) for v in values]
     maxes = [max(component) for component in zip(*rgbs)]
     return cast(HexColor, rgb_to_hex(tuple(maxes)))
 def resolve(
         deviceType: DeviceClass,
         deviceAttr: DeviceAttr,
         values: List[VTUnion]) -> VTUnion:  # todo: return should be VT
 def resolve(deviceType: DeviceClass, deviceAttr: DeviceAttr, values: List[VTUnion]) -> VTUnion:  # todo: return should be VT
     """
     return one value to use for this attr, given a set of them that
     have come in simultaneously. len(values) >= 1.
     bug: some callers are passing a device instance for 1st arg
     """
     if len(values) == 1:
         return values[0]
     if deviceAttr == DeviceAttr(L9['color']):
         return _maxColor(cast(List[HexColor], values))
     # incomplete. how-to-resolve should be on the DeviceAttr defs in the graph.
     if deviceAttr in map(DeviceAttr, [L9['rx'], L9['ry'], L9['zoom'], L9['focus'], L9['iris']]):
         floatVals = []
         for v in values:
             if isinstance(v, Literal):
                 floatVals.append(float(v.toPython()))
             elif isinstance(v, (int, float)):
                 floatVals.append(float(v))
             else:
                 raise TypeError(repr(v))
         # averaging with zeros? not so good
         return sum(floatVals) / len(floatVals)
     return max(values)
 def toOutputAttrs(
         deviceType: DeviceClass,
         deviceAttrSettings: Dict[DeviceAttr, VTUnion  # TODO
         deviceAttrSettings: Dict[
             DeviceAttr,
             VTUnion  # TODO
                                 ]) -> Dict[OutputAttr, OutputValue]:
     return dict((OutputAttr(u), OutputValue(v)) for u, v in untype_toOutputAttrs(deviceType, deviceAttrSettings).items())
 def untype_toOutputAttrs(deviceType, deviceAttrSettings) -> Dict[URIRef, int]:
     """
     Given device attr settings like {L9['color']: Literal('#ff0000')},
     return a similar dict where the keys are output attrs (like
     L9['red']) and the values are suitable for Collector.setAttr
     :outputAttrRange happens before we get here.
     """
     def floatAttr(attr, default=0):
         out = deviceAttrSettings.get(attr)
         if out is None:
             return default
         return float(out.toPython()) if isinstance(out, Literal) else out
     def rgbAttr(attr):
         color = deviceAttrSettings.get(attr, '#000000')
         r, g, b = hex_to_rgb(color)
         return r, g, b
     def cmyAttr(attr):
         rgb = sRGBColor.new_from_rgb_hex(deviceAttrSettings.get(attr, '#000000'))
         out = colormath.color_conversions.convert_color(rgb, CMYColor)
         return (_8bit(out.cmy_c), _8bit(out.cmy_m), _8bit(out.cmy_y))
     def fine16Attr(attr, scale=1.0):
         x = floatAttr(attr) * scale
         hi = _8bit(x)
         lo = _8bit((x * 255) % 1.0)
         return hi, lo
     def choiceAttr(attr):
         # todo
         if deviceAttrSettings.get(attr) == L9['g1']:
             return 3
         if deviceAttrSettings.get(attr) == L9['g2']:
             return 10
         return 0
     if deviceType == L9['ChauvetColorStrip']:
         r, g, b = rgbAttr(L9['color'])
         return {L9['mode']: 215, L9['red']: r, L9['green']: g, L9['blue']: b}
     elif deviceType == L9['Bar612601d']:
         r, g, b = rgbAttr(L9['color'])
         return {L9['red']: r, L9['green']: g, L9['blue']: b}
     elif deviceType == L9['LedPar90']:
         r, g, b = rgbAttr(L9['color'])
         w = _8bit(floatAttr(L9['white']))
         return {L9['master']: 255, L9['red']: r, L9['green']: g, L9['blue']: b, L9['white']: w}
     elif deviceType == L9['LedPar54']:
         r, g, b = rgbAttr(L9['color'])
         w = _8bit(floatAttr(L9['white']))
         return {L9['master']: 255, L9['red']: r, L9['green']: g, L9['blue']: b, L9['white']: w, L9['strobe']: 0}
     elif deviceType == L9['SimpleDimmer']:
         return {L9['level']: _8bit(floatAttr(L9['brightness']))}
     elif deviceType == L9['MegaFlash']:
         return {
             L9['brightness']: _8bit(floatAttr(L9['brightness'])),
             L9['strobeSpeed']: _8bit(floatAttr(L9['strobeSpeed'])),
+        }
     elif deviceType == L9['Mini15']:
         out = {
             L9['rotationSpeed']: 0,  # seems to have no effect
             L9['dimmer']: 255,
             L9['colorChange']: 0,
             L9['colorSpeed']: 0,
             L9['goboShake']: _8bit(floatAttr(L9['goboShake'])),
+        }
         out[L9['goboChoose']] = {
             L9['open']: 0,
             L9['mini15Gobo1']: 10,
             L9['mini15Gobo2']: 20,
             L9['mini15Gobo3']: 30,
         }[deviceAttrSettings.get(L9['mini15GoboChoice'], L9['open'])]
         out[L9['red']], out[L9['green']], out[L9['blue']] = rgbAttr(L9['color'])
         out[L9['xRotation']], out[L9['xFine']] = fine16Attr(L9['rx'], 1 / 540)
         out[L9['yRotation']], out[L9['yFine']] = fine16Attr(L9['ry'], 1 / 240)
         # didn't find docs on this, but from tests it looks like 64 fine steps takes you to the next coarse step
         return out
     elif deviceType == L9['ChauvetHex12']:
         out = {}
         out[L9['red']], out[L9['green']], out[L9['blue']] = r, g, b = rgbAttr(L9['color'])
         out[L9['amber']] = 0
         out[L9['white']] = min(r, g, b)
         out[L9['uv']] = _8bit(floatAttr(L9['uv']))
         return out
     elif deviceType == L9['Source4LedSeries2']:
         out = {}
         out[L9['red']], out[L9['green']], out[L9['blue']] = rgbAttr(L9['color'])

src/light9/collector/weblisteners.py

➞

Show inline comments

 import asyncio
 import io
 import json
 import logging
 import time
 from typing import Any, Awaitable, Dict, List, Protocol, Tuple
 import fastavro
 from fastavro.schema import load_schema
 from light9.collector.output import Output as OutputInstance
 from light9.newtypes import (DeviceUri, DmxIndex, DmxMessageIndex, OutputAttr, OutputUri, OutputValue)
 import starlette.websockets
 import websockets
 log = logging.getLogger('weblisteners')
 def shortenOutput(out: OutputUri) -> str:
     return str(out).rstrip('/').rsplit('/', 1)[-1]
 class UiListener(Protocol):
     async def sendMessage(self, msg):
         ...
 class WebListeners:
     def __init__(self) -> None:
         self.CollectorUpdateSchema = load_schema('avro/CollectorUpdate.avsc')
         self.clients: List[Tuple[UiListener, Dict[DeviceUri, Dict[OutputAttr, OutputValue]]]] = []
         self.pendingMessageForDev: Dict[DeviceUri, Tuple[Dict[OutputAttr, OutputValue], Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri,
                                                                                                                                  DmxMessageIndex]]]] = {}
         self.lastFlush = 0
         asyncio.create_task(self.flusher())
     def addClient(self, client: UiListener):
         self.clients.append((client, {}))  # seen = {dev: attrs}
         log.info('added client %s %s', len(self.clients), client)
         # todo: it would be nice to immediately fill in the client on the
         # latest settings, but I lost them so I can't.
     def delClient(self, client: UiListener):
         self.clients = [(c, t) for c, t in self.clients if c != client]
         log.info('delClient %s, %s left', client, len(self.clients))
     def outputAttrsSet(self, dev: DeviceUri, attrs: Dict[OutputAttr, Any], outputMap: Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]]):
         """called often- don't be slow"""
         self.pendingMessageForDev[dev] = (attrs, outputMap)
         # maybe put on a stack for flusher or something
     async def flusher(self):
         while True:
             try:
                 await self._flush()
             except (starlette.websockets.WebSocketDisconnect, websockets.exceptions.ConnectionClosed):
                 pass
             await asyncio.sleep(.02)
     async def _flush(self):
         now = time.time()
         if now < self.lastFlush + .02 or not self.clients:
             return
         self.lastFlush = now
         while self.pendingMessageForDev:
             dev, (attrs, outputMap) = self.pendingMessageForDev.popitem()
             msg = None  # lazy, since makeMsg is slow
             sendAwaits: List[Awaitable[None]] = []
             # this omits repeats, but can still send many
             # messages/sec. Not sure if piling up messages for the browser
             # could lead to slowdowns in the real dmx output.
             for client, seen in self.clients:
                 if seen.get(dev) == attrs:
                     continue
                 if msg is None:
                     msg = self.makeMsg(dev, attrs, outputMap)
                 seen[dev] = attrs
                 sendAwaits.append(client.sendMessage(msg))
             await asyncio.gather(*sendAwaits)
     def makeMsg(self, dev: DeviceUri, attrs: Dict[OutputAttr, Any], outputMap: Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]]):
         attrRows = []
         for attr, val in attrs.items():
             outputUri, bufIndex = outputMap[(dev, attr)]
             dmxIndex = DmxIndex(bufIndex + 1)
             attrRows.append({'attr': attr.rsplit('/')[-1], 'val': val, 'chan': (shortenOutput(outputUri), dmxIndex)})
         attrRows.sort(key=lambda r: r['chan'])
         for row in attrRows:
             row['chan'] = '%s %s' % (row['chan'][0], row['chan'][1])
         out = io.BytesIO()
         fastavro.schemaless_writer(out, self.CollectorUpdateSchema, {'OutputAttrsSet': {'dev': dev, 'attrs': attrRows}})
         msg = out.getvalue()
         log.info(f'made update message {len(msg)=}')
         return msg

src/light9/effect/edit.py

➞

Show inline comments

 from rdflib import URIRef, Literal
 import treq
 from rdfdb.patch import Patch
 from rdflib import Literal, URIRef
 from twisted.internet.defer import inlineCallbacks, returnValue
 import treq
 from light9 import networking
 from light9.curvecalc.curve import CurveResource
 from light9.namespaces import L9, RDF, RDFS
 from rdfdb.patch import Patch
 def clamp(x, lo, hi):
     return max(lo, min(hi, x))
 @inlineCallbacks
 def getMusicStatus():
     resp = yield treq.get(networking.musicPlayer.path('time'), timeout=.5)
     body = yield resp.json_content()
     returnValue(body)
 @inlineCallbacks
 def songEffectPatch(graph, dropped, song, event, ctx):
     """
     some uri was 'dropped' in the curvecalc timeline. event is 'default' or 'start' or 'end'.
     """
     with graph.currentState(tripleFilter=(dropped, None, None)) as g:
         droppedTypes = list(g.objects(dropped, RDF.type))
         droppedLabel = g.label(dropped)
         droppedCodes = list(g.objects(dropped, L9['code']))
     quads = []
     fade = 2 if event == 'default' else 0
     if _songHasEffect(graph, song, dropped):
         # bump the existing curve
         pass
     else:
         effect, q = _newEffect(graph, song, ctx)
         quads.extend(q)
         curve = graph.sequentialUri(song + "/curve-")
         yield _newEnvelopeCurve(graph, ctx, curve, droppedLabel, fade)
         quads.extend([
             (song, L9['curve'], curve, ctx),
             (effect, RDFS.label, droppedLabel, ctx),
             (effect, L9['code'], Literal('env = %s' % curve.n3()), ctx),
         ])
         if L9['EffectClass'] in droppedTypes:
             quads.extend([
                 (effect, RDF.type, dropped, ctx),
             ] + [(effect, L9['code'], c, ctx) for c in droppedCodes])
         elif L9['Submaster'] in droppedTypes:
             quads.extend([
                 (effect, L9['code'], Literal('out = %s * env' % dropped.n3()),
                  ctx),
                 (effect, L9['code'], Literal('out = %s * env' % dropped.n3()), ctx),
             ])
         else:
             raise NotImplementedError(
                 "don't know how to add an effect from %r (types=%r)" %
                 (dropped, droppedTypes))
             raise NotImplementedError("don't know how to add an effect from %r (types=%r)" % (dropped, droppedTypes))
         _maybeAddMusicLine(quads, effect, song, ctx)
     print("adding")
     for qq in quads:
         print(qq)
     returnValue(Patch(addQuads=quads))
 @inlineCallbacks
 def songNotePatch(graph, dropped, song, event, ctx, note=None):
     """
     drop into effectsequencer timeline
     ported from timeline.coffee makeNewNote
     """
     with graph.currentState(tripleFilter=(dropped, None, None)) as g:
         droppedTypes = list(g.objects(dropped, RDF.type))
     quads = []
     fade = 2 if event == 'default' else 0.1
     if note:
         musicStatus = yield getMusicStatus()
         songTime = musicStatus['t']
         _finishCurve(graph, note, quads, ctx, songTime)
     else:
         if L9['Effect'] in droppedTypes:
             musicStatus = yield getMusicStatus()
             songTime = musicStatus['t']
             note = _makeNote(graph, song, note, quads, ctx, dropped, songTime,
                              event, fade)
             note = _makeNote(graph, song, note, quads, ctx, dropped, songTime, event, fade)
         else:
             raise NotImplementedError
     returnValue((note, Patch(addQuads=quads)))
 def _point(ctx, uri, t, v):
     return [(uri, L9['time'], Literal(round(t, 3)), ctx),
             (uri, L9['value'], Literal(round(v, 3)), ctx)]
     return [(uri, L9['time'], Literal(round(t, 3)), ctx), (uri, L9['value'], Literal(round(v, 3)), ctx)]
 def _finishCurve(graph, note, quads, ctx, songTime):
     with graph.currentState() as g:
         origin = g.value(note, L9['originTime']).toPython()
         curve = g.value(note, L9['curve'])
     pt2 = graph.sequentialUri(curve + 'p')
     pt3 = graph.sequentialUri(curve + 'p')
     quads.extend([(curve, L9['point'], pt2, ctx)] +
                  _point(ctx, pt2, songTime - origin, 1) +
                  [(curve, L9['point'], pt3, ctx)] +
     quads.extend([(curve, L9['point'], pt2, ctx)] + _point(ctx, pt2, songTime - origin, 1) + [(curve, L9['point'], pt3, ctx)] +
                  _point(ctx, pt3, songTime - origin + .5, 0))
 def _makeNote(graph, song, note, quads, ctx, dropped, songTime, event, fade):
     note = graph.sequentialUri(song + '/n')
     curve = graph.sequentialUri(note + 'c')
     quads.extend([
         (song, L9['note'], note, ctx),
         (note, RDF.type, L9['Note'], ctx),
         (note, L9['curve'], curve, ctx),
         (note, L9['effectClass'], dropped, ctx),
         (note, L9['originTime'], Literal(songTime), ctx),
         (curve, RDF.type, L9['Curve'], ctx),
         (curve, L9['attr'], L9['strength'], ctx),
     ])
     if event == 'default':
         coords = [(0 - fade, 0), (0, 1), (20, 1), (20 + fade, 0)]
     elif event == 'start':
         coords = [
             (0 - fade, 0),
             (0, 1),
+        ]
     elif event == 'end':  # probably unused- goes to _finishCurve instead
         coords = [(20, 1), (20 + fade, 0)]
     else:
         raise NotImplementedError(event)
     for t, v in coords:
         pt = graph.sequentialUri(curve + 'p')
         quads.extend([(curve, L9['point'], pt, ctx)] + _point(ctx, pt, t, v))
     return note
 def _songHasEffect(graph, song, uri):
     """does this song have an effect of class uri or a sub curve for sub
     uri? this should be simpler to look up."""
     return False  # todo
 def musicCurveForSong(uri):
     return URIRef(uri + 'music')
 def _newEffect(graph, song, ctx):
     effect = graph.sequentialUri(song + "/effect-")
     quads = [
         (song, L9['effect'], effect, ctx),
         (effect, RDF.type, L9['Effect'], ctx),
+    ]
     print("_newEffect", effect, quads)
     return effect, quads
 @inlineCallbacks
 def _newEnvelopeCurve(graph, ctx, uri, label, fade=2):
     """this does its own patch to the graph"""
     cr = CurveResource(graph, uri)
     cr.newCurve(ctx, label=Literal(label))
     yield _insertEnvelopePoints(cr.curve, fade)
     cr.saveCurve()
 @inlineCallbacks
 def _insertEnvelopePoints(curve, fade=2):
     # wrong: we might not be adding to the currently-playing song.
     musicStatus = yield getMusicStatus()
     songTime = musicStatus['t']
     songDuration = musicStatus['duration']
     t1 = clamp(songTime - fade, .1, songDuration - .1 * 2) + fade
     t2 = clamp(songTime + 20, t1 + .1, songDuration)
     curve.insert_pt((t1 - fade, 0))
     curve.insert_pt((t1, 1))
     curve.insert_pt((t2, 1))
     curve.insert_pt((t2 + fade, 0))
 def _maybeAddMusicLine(quads, effect, song, ctx):
     """
     add a line getting the current music into 'music' if any code might
     be mentioning that var
     """
     for spoc in quads:
         if spoc[1] == L9['code'] and 'music' in spoc[2]:
             quads.extend([(effect, L9['code'],
                            Literal('music = %s' % musicCurveForSong(song).n3()),
                            ctx)])
             quads.extend([(effect, L9['code'], Literal('music = %s' % musicCurveForSong(song).n3()), ctx)])
             break

src/light9/effect/effect_functions.py

➞

Show inline comments

 import logging
 import random
 from typing import cast
 from PIL import Image
 from webcolors import rgb_to_hex
 from light9.effect.scale import scale
 from light9.effect.settings import DeviceSettings
 from light9.namespaces import L9
 from light9.newtypes import HexColor
 random.seed(0)
 log = logging.getLogger('effectfunc')
 def sample8(img, x, y, repeat=False):
+def sample8(img, x, y, repeat=False) -> tuple[int, int, int]:
     if not (0 <= y < img.height):
         return (0, 0, 0)
     if 0 <= x < img.width:
         return img.getpixel((x, y))
     elif not repeat:
         return (0, 0, 0)
     else:
         return img.getpixel((x % img.width, y))
 def effect_scale(strength: float, devs: DeviceSettings) -> DeviceSettings:
     out = []
     if strength != 0:
         for d, da, v in devs.asList():
             out.append((d, da, scale(v, strength)))
     return DeviceSettings(devs.graph, out)
 def effect_strobe(
         songTime: float,  #
         strength: float,
         period: float,
         onTime: float,
         devs: DeviceSettings) -> DeviceSettings:
     if period == 0:
         scl = 0
     else:
         scl = strength if (songTime % period) < onTime else 0
     return effect_scale(scl, devs)
 def effect_image(
     songTime: float,  #
     strength: float,
     period: float,
     image: Image.Image,
     devs: DeviceSettings,
 ) -> DeviceSettings:
     x = int((songTime / period) * image.width)
     out = []
-    for y, (d, da, v) in enumerate(devs.asOrderedList()):
+    for y, (d, da, v) in enumerate(devs.asList()):
         if da != L9['color']:
             continue
         color8 = sample8(image, x, y, repeat=True)
         color = rgb_to_hex(tuple(color8))
         out.append((d, da, scale(color, strength * v)))
     return DeviceSettings(devs.graph, out)
@@ \ No newline at end of file @@
         color = HexColor(rgb_to_hex(color8))
         out.append((d, da, scale(color, strength * cast(float, v))))
     return DeviceSettings(devs.graph, out)

src/light9/effect/effecteval2.py

➞

Show inline comments

 import traceback
 import inspect
 import logging
 import traceback
 from dataclasses import dataclass
 from typing import Callable, List, Optional
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import RDF
 from rdflib.term import Node
 from light9.effect.effect_function_library import EffectFunctionLibrary
 from light9.effect.settings import DeviceSettings, EffectSettings
 from light9.namespaces import L9
 from light9.newtypes import (DeviceAttr, DeviceUri, EffectAttr, EffectFunction, EffectUri, VTUnion)
 from light9.newtypes import (
     DeviceAttr,
     DeviceUri,
     EffectAttr,
     EffectFunction,
     EffectUri,
     VTUnion,
+)
 from light9.typedgraph import typedValue
 log = logging.getLogger('effecteval')
 @dataclass
 class Config:
     effectFunction: EffectFunction
     esettings: EffectSettings
     devSettings: Optional[DeviceSettings]  # the EffectSettings :effectAttr :devSettings item, if there was one
     func: Callable
     funcArgs: List[inspect.Parameter]
 @dataclass
 class EffectEval2:
     """Runs one effect code to turn EffectSettings (e.g. strength) into DeviceSettings"""
     graph: SyncedGraph
     uri: EffectUri
     lib: EffectFunctionLibrary
     config: Optional[Config] = None
     def __post_init__(self):
         self.graph.addHandler(self._compile)
     def _compile(self):
         self.config = None
         if not self.graph.contains((self.uri, RDF.type, L9['Effect'])):
             return
         try:
             effectFunction = typedValue(EffectFunction, self.graph, self.uri, L9['effectFunction'])
             effSets = []
             devSettings = None
             for s in self.graph.objects(self.uri, L9['setting']):
                 attr = typedValue(EffectAttr, self.graph, s, L9['effectAttr'])
                 if attr == L9['deviceSettings']:
                     value = typedValue(Node, self.graph, s, L9['value'])
                     rows = []
                     for ds in self.graph.objects(value, L9['setting']):
                         d = typedValue(DeviceUri, self.graph, ds, L9['device'])
                         da = typedValue(DeviceAttr, self.graph, ds, L9['deviceAttr'])
                         v = typedValue(VTUnion, self.graph, ds, L9['value'])
                         rows.append((d, da, v))
                     devSettings = DeviceSettings(self.graph, rows)
                 else:
                     value = typedValue(VTUnion, self.graph, s, L9['value'])
                     effSets.append((self.uri, attr, value))
             esettings = EffectSettings(self.graph, effSets)
             try:
                 effectFunction = typedValue(EffectFunction, self.graph, self.uri, L9['effectFunction'])
             except ValueError:
                 raise ValueError(f'{self.uri} has no :effectFunction')
             func = self.lib.getFunc(effectFunction)
             # This should be in EffectFunctionLibrary
             funcArgs = list(inspect.signature(func).parameters.values())
             self.config = Config(effectFunction, esettings, devSettings, func, funcArgs)
         except Exception:
             log.error(f"while compiling {self.uri}")
             traceback.print_exc()
     def compute(self, songTime: float, inputs: EffectSettings) -> DeviceSettings:
         """
         calls our function using inputs (publishedAttr attrs, e.g. :strength)
         and effect-level settings including a special attr called :deviceSettings
         with DeviceSettings as its value
         """
         if self.config is None:
             return DeviceSettings(self.graph, [])
         c = self.config
         kw = {}
         for arg in c.funcArgs:
             if arg.annotation == DeviceSettings:
                 v = c.devSettings
                 if v is None: # asked for ds but we have none
                     log.debug("%s asked for devs but we have none in config", self.uri)
                     return DeviceSettings(self.graph, [])
             elif arg.name == 'songTime':
                 v = songTime
             else:
                 eaForName = EffectAttr(L9[arg.name])
                 v = self._getEffectAttrValue(eaForName, inputs)
             kw[arg.name] = v
         if False and log.isEnabledFor(logging.DEBUG):
             log.debug('calling %s with %s', c.func, kw)
         return c.func(**kw)
     def _getEffectAttrValue(self, attr: EffectAttr, inputs: EffectSettings) -> VTUnion:

src/light9/effect/scale.py

➞

Show inline comments

 import logging
 from decimal import Decimal
 from webcolors import hex_to_rgb, rgb_to_hex
 from light9.newtypes import VTUnion
 log = logging.getLogger('scale')
 def scale(value: VTUnion, strength: float):
     if isinstance(value, Decimal):
         raise TypeError()
     if isinstance(value, str):
         if value[0] == '#':
             if strength == '#ffffff':
                 return value
             r, g, b = hex_to_rgb(value)
             # if isinstance(strength, Literal):
             #     strength = strength.toPython()
             # if isinstance(strength, str):
             #     sr, sg, sb = [v / 255 for v in hex_to_rgb(strength)]
             if True:
                 sr = sg = sb = strength
             return rgb_to_hex((int(r * sr), int(g * sg), int(b * sb)))
     elif isinstance(value, (int, float)):
         return value * strength
     raise NotImplementedError("%r,%r" % (value, strength))

src/light9/effect/sequencer/eval_faders.py

➞

Show inline comments

 import traceback
 import logging
 import time
 import traceback
 from dataclasses import dataclass
 from typing import List, Optional, cast
 from prometheus_client import Summary
 from rdfdb import SyncedGraph
 from rdflib import URIRef
 from rdflib.term import Node
 from light9.effect.effect_function_library import EffectFunctionLibrary
 from light9.effect.effecteval2 import EffectEval2
 from light9.effect.settings import DeviceSettings, EffectSettings
 from light9.namespaces import L9, RDF
 from light9.newtypes import EffectAttr, EffectUri, UnixTime
 from light9.typedgraph import typedValue
 log = logging.getLogger('seq.fader')
 COMPILE = Summary('compile_graph_fader', 'compile')
 COMPUTE_ALL_FADERS = Summary('compute_all_faders', 'compile')
 @dataclass
 class Fader:
     graph: SyncedGraph
     lib: EffectFunctionLibrary
     uri: URIRef
     effect: EffectUri
     setEffectAttr: EffectAttr
     value: Optional[float] = None  # mutable
     def __post_init__(self):
         self.ee = EffectEval2(self.graph, self.effect, self.lib)
 class FaderEval:
     """peer to Sequencer, but this one takes the current :Fader settings -> sendToCollector
     """
     def __init__(self, graph: SyncedGraph, lib: EffectFunctionLibrary):
         self.graph = graph
         self.lib = lib
         self.faders: List[Fader] = []
         self.grandMaster = 1.0
         self.graph.addHandler(self._compile)
         self.graph.addHandler(self._compileGm)
     @COMPILE.time()
     def _compile(self) -> None:
         """rebuild our data from the graph"""
         self.faders = []
         for fader in cast(list[URIRef], self.graph.subjects(RDF.type, L9['Fader'])):
             try:
                 self.faders.append(self._compileFader(fader))
             except ValueError:
                 pass
         # this could go in a second, smaller addHandler call to avoid rebuilding Fader objs constantly
         for f in self.faders:
             f.value = None
             try:
                 setting = typedValue(Node, self.graph, f.uri, L9['setting'])
             except ValueError:
                 continue
             try:
                 f.value = typedValue(float, self.graph, setting, L9['value'])
             except ValueError:
                 continue
     def _compileFader(self, fader: URIRef) -> Fader:
         effect = typedValue(EffectUri, self.graph, fader, L9['effect'])
         setting = typedValue(Node, self.graph, fader, L9['setting'])
         setAttr = typedValue(EffectAttr, self.graph, setting, L9['effectAttr'])
         return Fader(self.graph, self.lib, cast(URIRef, fader), effect, setAttr)
     def _compileGm(self):
         try:
             self.grandMaster = typedValue(float, self.graph, L9.grandMaster, L9.value)
         except ValueError:
             return
     @COMPUTE_ALL_FADERS.time()
     def computeOutput(self) -> DeviceSettings:
         faderEffectOutputs: List[DeviceSettings] = []
         now = UnixTime(time.time())
         for f in self.faders:
             try:
                 if f.value is None:
                     log.warning(f'{f.value=}; should be set during _compile. Skipping {f.uri}')
                     continue
                 v = f.value
                 v *= self.grandMaster
                 effectSettings = EffectSettings(self.graph, [(f.effect, f.setEffectAttr, v)])

src/light9/effect/sequencer/service.py

➞

Show inline comments

 """
 plays back effect notes from the timeline (and an untimed note from the faders)
 """
 import asyncio
 import json
 import logging
 import time
 from louie import dispatcher
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from sse_starlette.sse import EventSourceResponse
 from starlette.applications import Starlette
 from starlette.routing import Route
 from starlette_exporter import PrometheusMiddleware, handle_metrics
 from light9 import networking
 from light9.background_loop import loop_forever
 from light9 import networking
 from light9.collector.collector_client_asyncio import sendToCollector
 from light9.effect.effect_function_library import EffectFunctionLibrary
 from light9.effect.sequencer.eval_faders import FaderEval
 from light9.effect.sequencer.sequencer import Sequencer, StateUpdate
 from light9.run_local import log
 RATE = 20
 async def changes():
     state = {}
     q = asyncio.Queue()
     def onBroadcast(update):
         state.update(update)
         q.put_nowait(None)
     dispatcher.connect(onBroadcast, StateUpdate)
     lastSend = 0
     while True:
         await q.get()
         now = time.time()
         if now > lastSend + .2:
             lastSend = now
             yield json.dumps(state)
 async def send_page_updates(request):
     return EventSourceResponse(changes())
 def main():
     graph = SyncedGraph(networking.rdfdb.url, "effectSequencer")
     logging.getLogger('sse_starlette.sse').setLevel(logging.INFO)
     logging.getLogger('autodepgraphapi').setLevel(logging.INFO)
     logging.getLogger('syncedgraph').setLevel(logging.INFO)
     logging.getLogger('effecteval').setLevel(logging.INFO)
     logging.getLogger('seq.fader').setLevel(logging.INFO)
     # seq = Sequencer(graph, send)  # per-song timed notes
     lib = EffectFunctionLibrary(graph)
     faders = FaderEval(graph, lib)  # bin/fade's untimed effects
     #@metrics('computeAndSend').time() # needs rework with async
     async def update(first_run):
         ds = faders.computeOutput()
         await sendToCollector('effectSequencer', session='0', settings=ds)
     faders_loop = loop_forever(func=update, metric_prefix='faders', sleep_period=1 / RATE)
     app = Starlette(
         debug=True,
         routes=[
             Route('/updates', endpoint=send_page_updates),
         ],
+    )
     app.add_middleware(PrometheusMiddleware)
     app.add_route("/metrics", handle_metrics)
     return app
 app = main()

src/light9/effect/settings.py

➞

Show inline comments

 """
 Data structure and convertors for a table of (device,attr,value)
 rows. These might be effect attrs ('strength'), device attrs ('rx'),
 or output attrs (dmx channel).
 BareSettings means (attr,value), no device.
 """
 from __future__ import annotations
 import decimal
 import logging
 from dataclasses import dataclass
 from typing import Any, Dict, Iterable, List, Sequence, Set, Tuple, cast
 import numpy
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import Literal, URIRef
 from light9.collector.device import resolve
 from light9.localsyncedgraph import LocalSyncedGraph
 from light9.namespaces import L9, RDF
-from light9.newtypes import (DeviceAttr, DeviceUri, EffectAttr, HexColor, VTUnion)
 from light9.newtypes import DeviceAttr, DeviceUri, EffectAttr, HexColor, VTUnion
 log = logging.getLogger('settings')
 def parseHex(h):
     if h[0] != '#':
         raise ValueError(h)
     return [int(h[i:i + 2], 16) for i in (1, 3, 5)]
 def parseHexNorm(h):
     return [x / 255 for x in parseHex(h)]
 def toHex(rgbFloat: Sequence[float]) -> HexColor:
     assert len(rgbFloat) == 3
     scaled = (max(0, min(255, int(v * 255))) for v in rgbFloat)
     return HexColor('#%02x%02x%02x' % tuple(scaled))
 def getVal(graph, subj):
     lit = graph.value(subj, L9['value']) or graph.value(subj, L9['scaledValue'])
     ret = lit.toPython()
     if isinstance(ret, decimal.Decimal):
         ret = float(ret)
     return ret
 GraphType = SyncedGraph | LocalSyncedGraph
 class _Settings:
     """
     Generic for DeviceUri/DeviceAttr/VTUnion or EffectClass/EffectAttr/VTUnion
     default values are 0 or '#000000'. Internal rep must not store zeros or some
     comparisons will break.
     """
     EntityType = DeviceUri
     AttrType = DeviceAttr
     def __init__(self, graph: GraphType, settingsList: List[Tuple[Any, Any, VTUnion]]):
         self.graph = graph  # for looking up all possible attrs
         self._compiled: Dict[self.__class__.EntityType, Dict[self.__class__.AttrType, VTUnion]] = {}
         for e, a, v in settingsList:
             attrVals = self._compiled.setdefault(e, {})
             if a in attrVals:
                 v = resolve(
                     e,  # Hey, this is supposed to be DeviceClass (which is not convenient for us), but so far resolve() doesn't use that arg
                     a,
                     [attrVals[a], v])
             attrVals[a] = v
         # self._compiled may not be final yet- see _fromCompiled
         self._delZeros()
     @classmethod
     def _fromCompiled(cls, graph: GraphType, compiled: Dict[EntityType, Dict[AttrType, VTUnion]]):
         obj = cls(graph, [])
         obj._compiled = compiled
         obj._delZeros()
         return obj
     @classmethod
     def fromList(cls, graph: GraphType, others: List[_Settings]):
         """note that others may have multiple values for an attr"""
         self = cls(graph, [])
         for s in others:
             # if not isinstance(s, cls):
             #     raise TypeError(s)
             for row in s.asList():  # could work straight from s._compiled
                 if row[0] is None:
                     raise TypeError('bad row %r' % (row,))
                 dev, devAttr, value = row
                 devDict = self._compiled.setdefault(dev, {})
                 if devAttr in devDict:
                     existingVal: VTUnion = devDict[devAttr]
                     # raise NotImplementedError('fixme: dev is to be a deviceclass (but it is currently unused)')
                     value = resolve(dev, devAttr, [existingVal, value])
                 devDict[devAttr] = value
         self._delZeros()
         return self
     @classmethod
     def _mult(cls, weight, row, dd) -> VTUnion:
         if isinstance(row[2], str):
             prev = parseHexNorm(dd.get(row[1], '#000000'))
             return toHex(prev + weight * numpy.array(parseHexNorm(row[2])))
         else:
             return dd.get(row[1], 0) + weight * row[2]
     @classmethod
     def fromBlend(cls, graph: GraphType, others: List[Tuple[float, _Settings]]):
         """others is a list of (weight, Settings) pairs"""
         out = cls(graph, [])
         for weight, s in others:
             if not isinstance(s, cls):

0 comments (0 inline, 0 general)