technicalkiwi/lighting-controller
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: technicalkiwi:basic
Branches Tags
technicalkiwi:dev technicalkiwi:web technicalkiwi:basic technicalkiwi:ui technicalkiwi:pi technicalkiwi:full technicalkiwi:kat
technicalkiwi:1.0
..
compare: technicalkiwi:e5cf15d7b0a98a97b194536f0882c57b4a67e1a4
Branches Tags
technicalkiwi:web technicalkiwi:basic technicalkiwi:ui technicalkiwi:pi technicalkiwi:full technicalkiwi:kat technicalkiwi:dev
technicalkiwi:1.0

11 Commits
basic ... e5cf15d7b0

Author SHA1 Message Date
jimmy
e5cf15d7b0 Fix rainbow pattern step range in lighting controller 
- Change step calculation from beat_index % 2 to beat_index % 256
- Provides full 0-255 step range for rainbow pattern color cycling
- Fixes rainbow pattern that was limited to only 0 or 1 step values
- Alternating phase patterns still use % 2 for proper phase offset
 2025-09-19 01:29:08 +12:00
jimmy
c40b5629bf Fix pattern highlighting in lighting controller GUI 
- Add pattern name mapping to translate between MIDI handler names and GUI display names
- Fixes highlighting for patterns with underscores (sequential_pulse, alternating_phase, n_chase)
- Now properly highlights selected patterns in the button grid
 2025-09-19 01:15:29 +12:00
jimmy
a4a00021d8 Fix missing n3 attribute and async function issue 
- Add missing self.n3 = 1 attribute to MidiHandler initialization
- Fix update_rgb async function to properly yield control with await asyncio.sleep(0)
- Resolves TypeError about expecting coroutine but getting None
- Application now working properly with buttons and dials functional
 2025-09-19 00:07:37 +12:00
jimmy
f2dcdabf29 Fix indentation errors and reduce debug output 
- Comment out all debug logging statements to reduce console noise
- Fix empty if/else blocks by adding pass statements
- Remove beat logging, TCP server logging, and MIDI debug messages
- Keep only essential info, warning, and error messages
- Revert radiate delay separation back to using main delay parameter
 2025-09-19 00:02:51 +12:00
jimmy
5f7db51851 Add rate-limited parameter updates and message type system 
- Rate limit parameter updates to 100ms minimum interval
- Send immediate updates if rate limit allows, otherwise queue
- Process pending updates during beat handling
- All knob changes (CC30-37) now trigger parameter updates
- Add message type field: 'b' for beats, 'u' for updates
- Optimize message type to single letters to save packet space
- Prevents ESP-NOW network flooding during rapid knob adjustments
- All packets stay under 230-byte limit with automatic splitting
 2025-09-18 22:11:17 +12:00
jimmy
fcbe9e9094 Implement full parameter sending on pattern change and periodic updates 
- Send all parameters when pattern changes (may require 2 packets if >200 bytes)
- Send periodic parameter updates every 8 beats to keep bars synchronized
- Beat packets remain minimal for performance
- All packets stay under 230-byte limit
 2025-09-18 21:58:39 +12:00
jimmy
36dfda74b2 Update GUI layout and MIDI CC mappings: CC36=n3, CC37=delay, remove B1/B2 references 2025-09-18 20:35:31 +12:00
jimmy
8d0c9edf5d ws: adopt nested {'0': {...}} payloads 
midi: bind patterns to notes 36+; beat triggers selected pattern; include n index; CC map: 30=R 31=G 32=B 33=brightness 34=n1 35=n2 36=delay; send n1/n2 raw 0-127

gui: show n1 and n2 in status
 2025-09-17 20:22:11 +12:00
jimmy
1da2e30d4c midi: init read of CCs on startup (delay, brightness, RGB, beat enable); track bpm/pattern for GUI\nmain: integrate MidiHandler; add status panel for delay/brightness/RGB/pattern/BPM 2025-09-14 05:23:46 +12:00
jimmy
9ff38aa875 midi: add CC29 tempo reset, CC37 brightness; local beat flag; logging\nsound: add control server with RESET_TEMPO; logging; always send BPM 2025-09-14 04:53:24 +12:00
jimmy
3b869851b8 Add patterns 2025-09-09 21:40:27 +12:00
9 changed files with 1165 additions and 3337 deletions
Expand all files Collapse all files

2
Pipfile
View File

@@ -19,5 +19,5 @@ websocket-client = "*"
python_version = "3.12" python_version = "3.12"
[scripts] [scripts]
main = "python src/main.py" main = "python main.py"
dev = 'watchfiles "python src/main.py" src' dev = 'watchfiles "python src/main.py" src'

113
profiles/ring.json
View File

@@ -1,113 +0,0 @@
{
"tab_password": "",
"patterns": {
"on": {
"min_delay": 10,
"max_delay": 10000
},
"off": {
"min_delay": 10,
"max_delay": 10000
},
"rainbow": {
"Step Rate": "n1",
"min_delay": 10,
"max_delay": 10000
},
"transition": {
"min_delay": 10,
"max_delay": 10000
},
"chase": {
"Colour 1 Length": "n1",
"Colour 2 Length": "n2",
"Step 1": "n3",
"Step 2": "n4",
"min_delay": 10,
"max_delay": 10000
},
"pulse": {
"Attack": "n1",
"Hold": "n2",
"Decay": "n3",
"min_delay": 10,
"max_delay": 10000
},
"circle": {
"Head Rate": "n1",
"Max Length": "n2",
"Tail Rate": "n3",
"Min Length": "n4",
"min_delay": 10,
"max_delay": 10000
},
"blink": {
"min_delay": 10,
"max_delay": 10000
}
},
"lights": {
"ring1": {
"names": [
"dj"
],
"settings": {
"pattern": "on",
"brightness": 127,
"colors": [
"#000000"
],
"delay": 100,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"patterns": {
"on": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"ring2": {
"names": [
"ring2"
],
"settings": {
"pattern": "on",
"brightness": 127,
"colors": [
"#000000"
],
"delay": 100,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"patterns": {
"on": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
}
},
"tab_order": [
"ring1",
"ring2"
]
}

864
profiles/tt.json
View File

@@ -1,864 +0,0 @@
{
"lights": {
"sign": {
"names": [
"tt-sign",
"1"
],
"settings": {
"colors": [
"#968a00"
],
"brightness": 39,
"pattern": "circle",
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"pulse": {
"colors": [
"#ff00ff"
],
"delay": 657,
"n1": 100,
"n2": 10,
"n3": 100,
"n4": 10,
"n5": 10,
"n6": 10
},
"n_chase": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"rainbow": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"off": {
"colors": [
"#0000ff",
"#ff0000"
],
"delay": 10000,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"blink": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"transition": {
"colors": [
"#ff00ff",
"#ffff00"
],
"delay": 10000,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"circle": {
"colors": [
"#0000ff",
"#ff0000"
],
"delay": 10000,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"chase": {
"colors": [
"#000091",
"#00d800"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"dj": {
"names": [
"dj"
],
"settings": {
"colors": [
"#0000ff",
"#ff0000"
],
"brightness": 39,
"pattern": "transition",
"delay": 10000,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"rainbow": {
"colors": [
"#00006a"
],
"delay": 17,
"n1": 1,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"on": {
"colors": [
"#ff0062",
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"blink": {
"colors": [
"#0000d0"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"pulse": {
"delay": 1002,
"colors": [
"#006600",
"#0000ff"
],
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"transition": {
"colors": [
"#0000ff",
"#ff0000"
],
"delay": 10000,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10
},
"n_chase": {
"n1": 11,
"n2": 13,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"delay": 639,
"colors": [
"#0000ff"
]
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"circle": {
"colors": [
"#0001bd",
"#00ff00"
],
"delay": 1778,
"n1": 20,
"n2": 40,
"n3": 40,
"n4": 0
},
"chase": {
"colors": [
"#8d00ff",
"#ff0077"
],
"delay": 69,
"n1": 30,
"n2": 30,
"n3": 5,
"n4": 30
}
}
}
},
"middle": {
"names": [
"middle1",
"middle2",
"middle3",
"middle4"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 44,
"pattern": "on",
"delay": 520,
"patterns": {
"flicker": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"sides": {
"names": [
"left",
"right"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 13,
"pattern": "on",
"delay": 520,
"patterns": {
"on": {
"colors": [
"#ff00ff"
],
"delay": 988,
"n1": 100,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"pulse": {
"n1": 100,
"n2": 100,
"n3": 100,
"n4": 10,
"delay": 411,
"colors": [
"#ff00ff"
]
}
}
}
},
"outside": {
"names": [
"outside"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 76,
"pattern": "on",
"delay": 520,
"n1": -17,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"transition": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"middle1": {
"names": [
"middle1"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 59,
"pattern": "on",
"delay": 520,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"flicker": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"pulse": {
"delay": 1096,
"colors": [
"#0000ff"
],
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"rainbow": {
"n1": 1,
"n2": 10,
"n3": 10,
"n4": 10,
"delay": 2884,
"colors": [
"#000000"
]
},
"transition": {
"colors": [
"#0000ff",
"#ff0000"
],
"delay": 269,
"n1": 5,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"middle2": {
"names": [
"middle2"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 141,
"pattern": "on",
"delay": 520,
"patterns": {
"flicker": {
"colors": [
"#000078"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"pulse": {
"colors": [
"#0000a0",
"#720000"
],
"delay": 4102,
"n1": 100,
"n2": 10,
"n3": 100,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"middle3": {
"names": [
"middle3"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "on",
"delay": 520,
"patterns": {
"flicker": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"on": {
"colors": [
"#00c4a5"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"middle4": {
"names": [
"middle4"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "on",
"delay": 520,
"patterns": {
"flicker": {
"colors": [
"#ff00d6"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"front1": {
"names": [
"front1"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 255,
"pattern": "on",
"delay": 520,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"on": {
"colors": [
"#ff00ff",
"#0000ff"
],
"delay": 2409,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"pulse": {
"colors": [
"#000090"
],
"delay": 1051,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"transition": {
"colors": [
"#ff0000",
"#0000ff"
],
"delay": 2564,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"front2": {
"names": [
"front2"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 255,
"pattern": "off",
"delay": 520,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"on": {
"colors": [
"#ff00ff"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"rainbow": {
"colors": [
"#00006b"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"transition": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
},
"front3": {
"names": [
"front3"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 29,
"pattern": "on",
"delay": 520,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"patterns": {
"on": {
"colors": [
"#d200d1"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"off": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"rainbow": {
"colors": [
"#00006b"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
},
"transition": {
"colors": [
"#000000"
],
"delay": 99,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10
}
}
}
}
},
"tab_password": "qwerty1234",
"tab_order": [
"sign",
"dj",
"middle",
"sides",
"outside",
"middle1",
"middle2",
"middle3",
"middle4",
"front1",
"front2",
"front3"
]
}

192
settings.json
View File

@@ -1,50 +1,162 @@
{ {
"tab_password": "", "lights": {
"current_profile": "ring", "sign": {
"patterns": { "names": [
"on": { "tt-sign",
"min_delay": 10, "1"
"max_delay": 10000 ],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00"
],
"brightness": 9,
"pattern": "off",
"delay": 50
}
}, },
"off": { "dj": {
"min_delay": 10, "names": [
"max_delay": 10000 "dj"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
}, },
"rainbow": { "middle": {
"Step Rate": "n1", "names": [
"min_delay": 10, "middle1",
"max_delay": 10000 "middle2",
"middle3",
"middle4"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
}, },
"transition": { "sides": {
"min_delay": 10, "names": [
"max_delay": 10000 "left",
"right"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "on",
"delay": 520
}
}, },
"chase": { "outside": {
"Colour 1 Length": "n1", "names": [
"Colour 2 Length": "n2", "outside"
"Step 1": "n3", ],
"Step 2": "n4", "settings": {
"min_delay": 10, "colors": [
"max_delay": 10000 "#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "on",
"delay": 520
}
}, },
"pulse": { "middle1": {
"Attack": "n1", "names": [
"Hold": "n2", "middle1"
"Decay": "n3", ],
"min_delay": 10, "settings": {
"max_delay": 10000 "colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
}, },
"circle": { "middle2": {
"Head Rate": "n1", "names": [
"Max Length": "n2", "middle2"
"Tail Rate": "n3", ],
"Min Length": "n4", "settings": {
"min_delay": 10, "colors": [
"max_delay": 10000 "#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
}, },
"blink": { "middle3": {
"min_delay": 10, "names": [
"max_delay": 10000 "middle3"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
},
"middle4": {
"names": [
"middle4"
],
"settings": {
"colors": [
"#0000ff",
"#c30074",
"#00ff00",
"#000000"
],
"brightness": 6,
"pattern": "flicker",
"delay": 520
}
} }
} },
"patterns": [
"on",
"off",
"blink",
"rainbow_cycle",
"color_transition",
"theater_chase",
"flicker",
"pulse"
]
} }

38
src/bar_config.py Normal file
View File

@@ -0,0 +1,38 @@
# LED Bar Configuration
# Modify these names as needed for your setup
# LED Bar Names/IDs - 4 left bars + 4 right bars
LED_BAR_NAMES = [
"100", # Left Bar 1
"101", # Left Bar 2
"102", # Left Bar 3
"103", # Left Bar 4
"104", # Right Bar 1
"105", # Right Bar 2
"106", # Right Bar 3
"107", # Right Bar 4
]
# Left and right bar groups for spatial control
LEFT_BARS = ["100", "101", "102", "103"]
RIGHT_BARS = ["104", "105", "106", "107"]
# Number of LED bars
NUM_BARS = len(LED_BAR_NAMES)
# Default settings for all bars
DEFAULT_BAR_SETTINGS = {
"pattern": "pulse",
"delay": 100,
"colors": [(0, 255, 0)], # Default green
"brightness": 10,
"num_leds": 200,
"n1": 10,
"n2": 10,
"n3": 1,
"n": 0,
}
# ESP-NOW broadcast settings
ESP_NOW_CHANNEL = 1
ESP_NOW_ENCRYPTION = False

2309
src/main.py
View File

File diff suppressed because it is too large Load Diff

639
src/midi.py
View File

@@ -1,105 +1,570 @@
import mido import mido
import asyncio import asyncio
import time import networking
import networking # <--- This will now correctly import your module import socket
import json
import logging # Added logging import
import time # Added for initial state read
import tkinter as tk
from tkinter import ttk, messagebox # Import messagebox for confirmations
from bar_config import LED_BAR_NAMES, DEFAULT_BAR_SETTINGS
# Pattern name mapping for shorter JSON payloads
PATTERN_NAMES = {
"flicker": "f",
"fill_range": "fr",
"n_chase": "nc",
"alternating": "a",
"pulse": "p",
"rainbow": "r",
"specto": "s",
"radiate": "rd",
}
async def midi_to_websocket_listener(midi_port_index: int, websocket_uri: str): # Configure logging
""" DEBUG_MODE = True # Set to False for INFO level logging
Listens to a specific MIDI port and sends data to a WebSocket server logging.basicConfig(level=logging.DEBUG if DEBUG_MODE else logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
when Note 32 (and 33) is pressed.
"""
delay = 100 # Default delay value
# 1. Get MIDI port name # TCP Server Configuration
TCP_HOST = "127.0.0.1"
TCP_PORT = 65432
# Sound Control Server Configuration (for sending reset)
SOUND_CONTROL_HOST = "127.0.0.1"
SOUND_CONTROL_PORT = 65433
class MidiHandler:
def __init__(self, midi_port_index: int, websocket_uri: str):
self.midi_port_index = midi_port_index
self.websocket_uri = websocket_uri
self.ws_client = networking.WebSocketClient(websocket_uri)
self.delay = 100 # Default delay value, controlled by MIDI controller
self.brightness = 100 # Default brightness value, controlled by MIDI controller
self.tcp_host = TCP_HOST
self.tcp_port = TCP_PORT
self.beat_sending_enabled = True # New: Local flag for beat sending
self.sound_control_host = SOUND_CONTROL_HOST
self.sound_control_port = SOUND_CONTROL_PORT
# RGB controlled by CC 30/31/32 (default green)
self.color_r = 0
self.color_g = 255
self.color_b = 0
# Generic parameters controlled via CC
# Raw CC-driven parameters (0-127)
self.n1 = 10
self.n2 = 10
self.n3 = 1
# Additional knobs (CC38-45)
self.knob1 = 0
self.knob2 = 0
self.knob3 = 0
self.knob4 = 0
self.knob5 = 0
self.knob6 = 0
self.knob7 = 0
self.knob8 = 0
# Current state for GUI display
self.current_bpm: float | None = None
self.current_pattern: str = ""
self.beat_index: int = 0
# Rate limiting for parameter updates
self.last_param_update: float = 0.0
self.param_update_interval: float = 0.1 # 100ms minimum between updates
self.pending_param_update: bool = False
# Sequential pulse pattern state
self.sequential_pulse_enabled: bool = False
self.sequential_pulse_step: int = 0
def _current_color_rgb(self) -> tuple:
r = max(0, min(255, int(self.color_r)))
g = max(0, min(255, int(self.color_g)))
b = max(0, min(255, int(self.color_b)))
return (r, g, b)
async def _handle_sequential_pulse(self):
"""Handle sequential pulse pattern: each bar pulses for 1 beat, then next bar, mirrored"""
from bar_config import LEFT_BARS, RIGHT_BARS
# Calculate which bar should pulse based on beat (1 beat per bar)
bar_index = self.beat_index % 4 # 0-3, cycles every 4 beats
# Create minimal payload - defaults to off
payload = {
"d": { # Defaults - off for all bars
"t": "b", # Message type: beat
"pt": "o", # off
}
}
# Set specific bars to pulse
left_bar = LEFT_BARS[bar_index]
right_bar = RIGHT_BARS[bar_index]
payload[left_bar] = {"pt": "p"} # pulse
payload[right_bar] = {"pt": "p"} # pulse
# logging.debug(f"[Sequential Pulse] Beat {self.beat_index}, pulsing bars {left_bar} and {right_bar}")
await self.ws_client.send_data(payload)
async def _handle_alternating_phase(self):
"""Handle alternating pattern with phase offset: every second bar uses different step"""
from bar_config import LED_BAR_NAMES
# Create minimal payload - same n1/n2 for all bars
payload = {
"d": { # Defaults - pattern and n1/n2
"t": "b", # Message type: beat
"pt": "a", # alternating
"n1": self.n1,
"n2": self.n2,
"s": self.beat_index % 2, # Default step for in-phase bars
}
}
# Set step offset for every second bar (bars 101, 103, 105, 107)
swap_bars = ["101", "103", "105", "107"]
for bar_name in LED_BAR_NAMES:
if bar_name in swap_bars:
# Send step offset for out-of-phase bars
payload[bar_name] = {"s": (self.beat_index + 1) % 2}
else:
# In-phase bars use defaults (no override needed)
payload[bar_name] = {}
# logging.debug(f"[Alternating Phase] Beat {self.beat_index}, step offset for bars {swap_bars}")
await self.ws_client.send_data(payload)
async def _send_full_parameters(self):
"""Send all parameters to bars - may require multiple packets due to size limit"""
from bar_config import LED_BAR_NAMES
# Calculate packet size for full parameters
full_payload = {
"d": {
"t": "u", # Message type: update
"pt": PATTERN_NAMES.get(self.current_pattern, self.current_pattern),
"dl": self.delay,
"cl": [self._current_color_rgb()],
"br": self.brightness,
"n1": self.n1,
"n2": self.n2,
"n3": self.n3,
"s": self.beat_index % 256, # Use full range for rainbow patterns
}
}
# Estimate size: ~200 bytes for defaults + 8 bars * 2 bytes = ~216 bytes
# This should fit in one packet, but let's be safe
payload_size = len(str(full_payload))
if payload_size > 200: # Split into 2 packets if too large
# Packet 1: Pattern and timing parameters
payload1 = {
"d": {
"t": "u", # Message type: update
"pt": PATTERN_NAMES.get(self.current_pattern, self.current_pattern),
"dl": self.delay,
"br": self.brightness,
}
}
for bar_name in LED_BAR_NAMES:
payload1[bar_name] = {}
# Packet 2: Color and pattern parameters
payload2 = {
"d": {
"t": "u", # Message type: update
"cl": [self._current_color_rgb()],
"n1": self.n1,
"n2": self.n2,
"n3": self.n3,
"s": self.beat_index % 2, # Keep step small (0 or 1) for alternating patterns
}
}
for bar_name in LED_BAR_NAMES:
payload2[bar_name] = {}
# logging.debug(f"[Full Params] Sending in 2 packets due to size ({payload_size} bytes)")
await self.ws_client.send_data(payload1)
await asyncio.sleep(0.01) # Small delay between packets
await self.ws_client.send_data(payload2)
else:
# Single packet
for bar_name in LED_BAR_NAMES:
full_payload[bar_name] = {}
# logging.debug(f"[Full Params] Sending single packet ({payload_size} bytes)")
await self.ws_client.send_data(full_payload)
async def _request_param_update(self):
"""Request a parameter update with rate limiting"""
import time
current_time = time.time()
if current_time - self.last_param_update >= self.param_update_interval:
# Can send immediately
self.last_param_update = current_time
await self._send_full_parameters()
# logging.debug("[Rate Limit] Parameter update sent immediately")
else:
# Rate limited - mark as pending
self.pending_param_update = True
# logging.debug("[Rate Limit] Parameter update queued (rate limited)")
async def _send_normal_pattern(self):
"""Send normal pattern to all bars - include required parameters"""
# Patterns that need specific parameters
patterns_needing_params = ["alternating", "flicker", "n_chase", "rainbow", "radiate"]
payload = {
"d": { # Defaults
"t": "b", # Message type: beat
"pt": PATTERN_NAMES.get(self.current_pattern, self.current_pattern),
}
}
# Add required parameters for patterns that need them
if self.current_pattern in patterns_needing_params:
payload["d"].update({
"n1": self.n1,
"n2": self.n2,
"n3": self.n3,
"dl": self.delay,
"s": self.beat_index % 256, # Use full range for rainbow patterns
})
# Add empty entries for each bar (they'll use defaults)
for bar_name in LED_BAR_NAMES:
payload[bar_name] = {}
# logging.debug(f"[Beat] Triggering '{self.current_pattern}' for {len(LED_BAR_NAMES)} bars")
await self.ws_client.send_data(payload)
async def _send_reset_to_sound(self):
try:
reader, writer = await asyncio.open_connection(self.sound_control_host, self.sound_control_port)
cmd = "RESET_TEMPO\n".encode('utf-8')
writer.write(cmd)
await writer.drain()
resp = await reader.read(100)
logging.info(f"[MidiHandler - Control] Sent RESET_TEMPO, response: {resp.decode().strip()}")
writer.close()
await writer.wait_closed()
except Exception as e:
logging.error(f"[MidiHandler - Control] Failed to send RESET_TEMPO: {e}")
async def _handle_tcp_client(self, reader, writer):
addr = writer.get_extra_info('peername')
logging.info(f"[MidiHandler - TCP Server] Connected by {addr}") # Changed to info
try:
while True:
data = await reader.read(4096) # Read up to 4KB of data
if not data:
logging.info(f"[MidiHandler - TCP Server] Client {addr} disconnected.") # Changed to info
break
message = data.decode().strip()
# logging.debug(f"[MidiHandler - TCP Server] Received from {addr}: {message}") # Changed to debug
if self.beat_sending_enabled:
try:
# Attempt to parse as float (BPM) from sound.py
bpm_value = float(message)
self.current_bpm = bpm_value
# On each beat, trigger currently selected pattern(s)
if not self.current_pattern:
pass # No pattern selected yet; ignoring beat
else:
self.beat_index = (self.beat_index + 1) % 1000000
# Send periodic parameter updates every 8 beats
if self.beat_index % 8 == 0:
await self._send_full_parameters()
# Check for pending parameter updates (rate limited)
if self.pending_param_update:
import time
current_time = time.time()
if current_time - self.last_param_update >= self.param_update_interval:
self.last_param_update = current_time
self.pending_param_update = False
await self._send_full_parameters()
# logging.debug("[Rate Limit] Pending parameter update sent")
if self.current_pattern == "sequential_pulse":
# Sequential pulse pattern: each bar pulses for 1 beat, then next bar, mirrored
await self._handle_sequential_pulse()
elif self.current_pattern == "alternating_phase":
# Alternating pattern with phase offset: every second bar is out of phase
await self._handle_alternating_phase()
elif self.current_pattern:
# Normal pattern mode - run on all bars
await self._send_normal_pattern()
except ValueError:
logging.warning(f"[MidiHandler - TCP Server] Received non-BPM message from {addr}, not forwarding: {message}") # Changed to warning
except Exception as e:
logging.error(f"[MidiHandler - TCP Server] Error processing received message from {addr}: {e}") # Changed to error
else:
pass # Beat sending disabled
except asyncio.CancelledError:
logging.info(f"[MidiHandler - TCP Server] Client handler for {addr} cancelled.") # Changed to info
except Exception as e:
logging.error(f"[MidiHandler - TCP Server] Error handling client {addr}: {e}") # Changed to error
finally:
logging.info(f"[MidiHandler - TCP Server] Closing connection for {addr}") # Changed to info
writer.close()
await writer.wait_closed()
async def _midi_tcp_server(self):
server = await asyncio.start_server(
lambda r, w: self._handle_tcp_client(r, w), self.tcp_host, self.tcp_port)
addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
logging.info(f"[MidiHandler - TCP Server] Serving on {addrs}") # Changed to info
async with server:
await server.serve_forever()
async def _read_initial_cc_state(self, port, timeout_s: float = 0.5):
"""Read initial CC values from the MIDI device for a short period to populate state."""
start = time.time()
while time.time() - start < timeout_s:
msg = port.receive(block=False)
if msg and msg.type == 'control_change':
if msg.control == 36:
self.n3 = max(1, msg.value)
logging.info(f"[Init] n3 set to {self.n3} from CC36")
elif msg.control == 37:
self.delay = msg.value * 4
logging.info(f"[Init] Delay set to {self.delay} ms from CC37")
elif msg.control == 39:
self.delay = msg.value * 4
logging.info(f"[Init] Delay set to {self.delay} ms from CC39")
elif msg.control == 33:
self.brightness = round((msg.value / 127) * 100)
logging.info(f"[Init] Brightness set to {self.brightness} from CC33")
elif msg.control == 30:
self.color_r = round((msg.value / 127) * 255)
logging.info(f"[Init] Red set to {self.color_r} from CC30")
elif msg.control == 31:
self.color_g = round((msg.value / 127) * 255)
logging.info(f"[Init] Green set to {self.color_g} from CC31")
elif msg.control == 32:
self.color_b = round((msg.value / 127) * 255)
logging.info(f"[Init] Blue set to {self.color_b} from CC32")
elif msg.control == 34:
self.n1 = int(msg.value)
logging.info(f"[Init] n1 set to {self.n1} from CC34")
elif msg.control == 35:
self.n2 = int(msg.value)
logging.info(f"[Init] n2 set to {self.n2} from CC35")
elif msg.control == 27:
self.beat_sending_enabled = (msg.value == 127)
logging.info(f"[Init] Beat sending {'ENABLED' if self.beat_sending_enabled else 'DISABLED'} from CC27")
await asyncio.sleep(0.001)
async def _midi_listener(self):
logging.info("Midi function") # Changed to info
"""
Listens to a specific MIDI port and sends data to a WebSocket server
when Note 32 (and 33) is pressed.
"""
# 1. Get MIDI port name
port_names = mido.get_input_names()
if not port_names:
logging.warning("No MIDI input ports found. Please connect your device.") # Changed to warning
return
if not (0 <= self.midi_port_index < len(port_names)):
logging.error(f"Error: MIDI port index {self.midi_port_index} out of range. Available ports: {port_names}") # Changed to error
logging.info("Available ports:") # Changed to info
for i, name in enumerate(port_names):
logging.info(f" {i}: {name}") # Changed to info
return
midi_port_name = port_names[self.midi_port_index]
logging.info(f"Selected MIDI input port: {midi_port_name}") # Changed to info
try:
with mido.open_input(midi_port_name) as port:
logging.info(f"MIDI port '{midi_port_name}' opened. Press Ctrl+C to stop.") # Changed to info
# Read initial controller state briefly
await self._read_initial_cc_state(port)
while True:
msg = port.receive(block=False) # Non-blocking read
if msg:
# logging.debug(msg) # Changed to debug
match msg.type:
case 'note_on':
# logging.debug(f" Note ON: Note={msg.note}, Velocity={msg.velocity}, Channel={msg.channel}") # Changed to debug
# Bank1 patterns starting at MIDI note 36
pattern_bindings: list[str] = [
# Pulse patterns (row 1)
"pulse",
"sequential_pulse",
# Alternating patterns (row 2)
"alternating",
"alternating_phase",
# Chase/movement patterns (row 3)
"n_chase",
"rainbow",
# Effect patterns (row 4)
"flicker",
"radiate",
]
idx = msg.note - 36
if 0 <= idx < len(pattern_bindings):
pattern_name = pattern_bindings[idx]
self.current_pattern = pattern_name
logging.info(f"[Select] Pattern selected via note {msg.note}: {self.current_pattern} (n1={self.n1}, n2={self.n2})")
# Send full parameters when pattern changes
await self._send_full_parameters()
else:
pass # Note not bound to patterns
case 'control_change':
match msg.control:
case 36:
self.n3 = max(1, msg.value) # Update n3 step rate
logging.info(f"n3 set to {self.n3} by MIDI controller (CC36)")
await self._request_param_update()
case 37:
self.delay = msg.value * 4 # Update instance delay
logging.info(f"Delay set to {self.delay} ms by MIDI controller (CC37)")
await self._request_param_update()
case 38:
self.n1 = msg.value # pulse n1 for pulse patterns
logging.info(f"Pulse n1 set to {self.n1} by MIDI controller (CC38)")
await self._request_param_update()
case 39:
self.n2 = msg.value # pulse n2 for pulse patterns
logging.info(f"Pulse n2 set to {self.n2} by MIDI controller (CC39)")
await self._request_param_update()
case 40:
self.n1 = msg.value # n1 for alternating patterns
logging.info(f"Alternating n1 set to {self.n1} by MIDI controller (CC40)")
await self._request_param_update()
case 41:
self.n2 = msg.value # n2 for alternating patterns
logging.info(f"Alternating n2 set to {self.n2} by MIDI controller (CC41)")
await self._request_param_update()
case 42:
self.n1 = msg.value # radiate n1 for radiate patterns
logging.info(f"Radiate n1 set to {self.n1} by MIDI controller (CC42)")
await self._request_param_update()
case 43:
self.delay = msg.value * 4 # delay for radiate patterns
logging.info(f"Delay set to {self.delay} ms by MIDI controller (CC43)")
await self._request_param_update()
case 44:
self.knob7 = msg.value
logging.info(f"Knob7 set to {self.knob7} by MIDI controller (CC44)")
await self._request_param_update()
case 45:
self.knob8 = msg.value
logging.info(f"Knob8 set to {self.knob8} by MIDI controller (CC45)")
await self._request_param_update()
case 27:
if msg.value == 127:
self.beat_sending_enabled = True
logging.info("[MidiHandler - Listener] Beat sending ENABLED by MIDI control.") # Changed to info
elif msg.value == 0:
self.beat_sending_enabled = False
logging.info("[MidiHandler - Listener] Beat sending DISABLED by MIDI control.") # Changed to info
case 29:
if msg.value == 127:
logging.info("[MidiHandler - Listener] RESET_TEMPO requested by control 29.")
await self._send_reset_to_sound()
case 33:
# Map 0-127 to 0-100 brightness scale
self.brightness = round((msg.value / 127) * 100)
logging.info(f"Brightness set to {self.brightness} by MIDI controller (CC33)")
await self._request_param_update()
case 30:
# Red 0-127 -> 0-255
self.color_r = round((msg.value / 127) * 255)
logging.info(f"Red set to {self.color_r}")
await self._request_param_update()
case 31:
# Green 0-127 -> 0-255
self.color_g = round((msg.value / 127) * 255)
logging.info(f"Green set to {self.color_g}")
await self._request_param_update()
case 32:
# Blue 0-127 -> 0-255
self.color_b = round((msg.value / 127) * 255)
logging.info(f"Blue set to {self.color_b}")
await self._request_param_update()
case 34:
self.n1 = int(msg.value)
logging.info(f"n1 set to {self.n1} by MIDI controller (CC34)")
await self._request_param_update()
case 35:
self.n2 = int(msg.value)
logging.info(f"n2 set to {self.n2} by MIDI controller (CC35)")
await self._request_param_update()
await asyncio.sleep(0.001) # Important: Yield control to asyncio event loop
except mido.PortsError as e:
logging.error(f"Error opening MIDI port '{midi_port_name}': {e}") # Changed to error
except asyncio.CancelledError:
logging.info(f"MIDI listener cancelled.") # Changed to info
except Exception as e:
logging.error(f"An unexpected error occurred in MIDI listener: {e}") # Changed to error
async def run(self):
try:
await self.ws_client.connect()
logging.info(f"[MidiHandler] WebSocket client connected to {self.ws_client.uri}") # Changed to info
# List available MIDI ports for debugging
print(f"Available MIDI input ports: {mido.get_input_names()}")
print(f"Trying to open MIDI port index {self.midi_port_index}")
await asyncio.gather(
self._midi_listener(),
self._midi_tcp_server()
)
except mido.PortsError as e:
logging.error(f"[MidiHandler] Error opening MIDI port: {e}") # Changed to error
print(f"MIDI Port Error: {e}")
print(f"Available MIDI ports: {mido.get_input_names()}")
print("Please check your MIDI device connection and port index")
except asyncio.CancelledError:
logging.info("[MidiHandler] Tasks cancelled due to program shutdown.") # Changed to info
except KeyboardInterrupt:
logging.info("\n[MidiHandler] Program interrupted by user.") # Changed to info
finally:
logging.info("[MidiHandler] Main program finished. Closing WebSocket client...") # Changed to info
await self.ws_client.close()
logging.info("[MidiHandler] WebSocket client closed.") # Changed to info
def print_midi_ports():
logging.info("\n--- Available MIDI Input Ports ---") # Changed to info
port_names = mido.get_input_names() port_names = mido.get_input_names()
if not port_names: if not port_names:
print("No MIDI input ports found. Please connect your device.") logging.warning("No MIDI input ports found.") # Changed to warning
return else:
if not (0 <= midi_port_index < len(port_names)):
print(f"Error: MIDI port index {midi_port_index} out of range. Available ports: {port_names}")
print("Available ports:")
for i, name in enumerate(port_names): for i, name in enumerate(port_names):
print(f" {i}: {name}") logging.info(f" {i}: {name}") # Changed to info
return logging.info("----------------------------------") # Changed to info
midi_port_name = port_names[midi_port_index]
print(f"Selected MIDI input port: {midi_port_name}")
# 2. Initialize WebSocket client (using your actual networking.py)
ws_client = networking.WebSocketClient(websocket_uri)
try:
# 3. Connect WebSocket
await ws_client.connect()
print(f"WebSocket client connected to {ws_client.uri}")
# 4. Open MIDI port and start listening loop
with mido.open_input(midi_port_name) as port:
print(f"MIDI port '{midi_port_name}' opened. Press Ctrl+C to stop.")
while True:
msg = port.receive(block=False) # Non-blocking read
if msg:
match msg.type:
case 'note_on':
print(f" Note ON: Note={msg.note}, Velocity={msg.velocity}, Channel={msg.channel}")
match msg.note:
case 32:
await ws_client.send_data({
"names": ["1"],
"settings": {
"pattern": "pulse",
"delay": delay,
"colors": ["#00ff00"],
"brightness": 100,
"num_leds": 200,
}
})
case 33:
await ws_client.send_data({
"names": ["2"],
"settings": {
"pattern": "chase",
"speed": 10,
"color": "#00FFFF",
}
})
case 'control_change':
match msg.control:
case 36:
delay = msg.value * 4
print(f"Delay set to {delay} ms")
await asyncio.sleep(0.001) # Important: Yield control to asyncio event loop
except mido.PortsError as e:
print(f"Error opening MIDI port '{midi_port_name}': {e}")
except asyncio.CancelledError:
print(f"MIDI listener cancelled.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
finally:
# 5. Disconnect WebSocket and clean up
# This assumes your WebSocketClient has a ._connected attribute or similar way to check state.
# If your client's disconnect method is safe to call even if not connected, you can simplify.
await ws_client.close()
print("MIDI listener stopped and cleaned up.")
async def main(): async def main():
print_midi_ports()
# --- Configuration --- # --- Configuration ---
MIDI_PORT_INDEX = 1 # <--- IMPORTANT: Change this to the correct index for your device MIDI_PORT_INDEX = 1 # <--- IMPORTANT: Change this to the correct index for your device
WEBSOCKET_SERVER_URI = "ws://192.168.4.1:80/ws" WEBSOCKET_SERVER_URI = "ws://192.168.4.1:80/ws"
# --- End Configuration --- # --- End Configuration ---
try: midi_handler = MidiHandler(MIDI_PORT_INDEX, WEBSOCKET_SERVER_URI)
await midi_to_websocket_listener(MIDI_PORT_INDEX, WEBSOCKET_SERVER_URI) await midi_handler.run()
except KeyboardInterrupt:
print("\nProgram interrupted by user.")
finally:
print("Main program finished.")
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(main()) asyncio.run(main())

67
src/settings.py
View File

@@ -1,67 +0,0 @@
import json
class Settings(dict):
SETTINGS_FILE = "settings.json"
def __init__(self):
super().__init__()
self.load() # Load settings from file during initialization
def save(self):
try:
# Create a copy without lights and tab_order (these belong in profiles, not settings.json)
# But keep patterns, tab_password, and current_profile
settings_to_save = {k: v for k, v in self.items() if k not in ["lights", "tab_order"]}
# Ensure patterns are always included if they exist
if "patterns" in self:
settings_to_save["patterns"] = self["patterns"]
j = json.dumps(settings_to_save, indent=4)
with open(self.SETTINGS_FILE, 'w') as file:
file.write(j)
print("Settings saved successfully.")
except Exception as e:
print(f"Error saving settings: {e}")
def load(self):
try:
with open(self.SETTINGS_FILE, 'r') as file:
loaded_settings = json.load(file)
self.update(loaded_settings)
# Ensure patterns exist (they should always be in settings.json)
if "patterns" not in self:
# Initialize with default patterns if missing
self["patterns"] = {
"on": {"min_delay": 10, "max_delay": 10000},
"off": {"min_delay": 10, "max_delay": 10000},
"rainbow": {"Step Rate": "n1", "min_delay": 10, "max_delay": 10000},
"transition": {"min_delay": 10, "max_delay": 10000},
"chase": {
"Colour 1 Length": "n1",
"Colour 2 Length": "n2",
"Step 1": "n3",
"Step 2": "n4",
"min_delay": 10,
"max_delay": 10000
},
"pulse": {
"Attack": "n1",
"Hold": "n2",
"Decay": "n3",
"min_delay": 10,
"max_delay": 10000
},
"circle": {
"Head Rate": "n1",
"Max Length": "n2",
"Tail Rate": "n3",
"Min Length": "n4",
"min_delay": 10,
"max_delay": 10000
},
"blink": {"min_delay": 10, "max_delay": 10000}
}
self.save() # Save to persist the default patterns
print("Settings loaded successfully.")
except Exception as e:
print(f"Error loading settings {e}")
self.save()

278
src/sound.py
View File

@@ -4,121 +4,207 @@ import pyaudio
import aubio import aubio
import numpy as np import numpy as np
from time import sleep from time import sleep
import websocket # pip install websocket-client
import json import json
import socket
import time
import logging # Added logging import
import asyncio # Re-added asyncio import
import threading # Added threading for control server
seconds = 10 # how long this script should run (if not using infinite loop) # Configure logging
DEBUG_MODE = True # Set to False for INFO level logging
logging.basicConfig(level=logging.DEBUG if DEBUG_MODE else logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
bufferSize = 512 # TCP Server Configuration (assuming midi.py runs this)
windowSizeMultiple = 2 # or 4 for higher accuracy, but more computational cost MIDI_TCP_HOST = "127.0.0.1"
MIDI_TCP_PORT = 65432
audioInputDeviceIndex = 7 # use 'arecord -l' to check available audio devices # Sound Control Server Configuration (for midi.py to control sound.py)
audioInputChannels = 1 SOUND_CONTROL_HOST = "127.0.0.1"
SOUND_CONTROL_PORT = 65433
pa = pyaudio.PyAudio() class SoundBeatDetector:
def __init__(self, tcp_host: str, tcp_port: int):
self.tcp_host = tcp_host
self.tcp_port = tcp_port
self.tcp_socket = None
self.connected_to_midi = False
self.reconnect_delay = 1 # seconds
# Note: beat_sending_enabled is not used in this simplified flow
print("Available audio input devices:") self.bufferSize = 512
info = pa.get_host_api_info_by_index(0) self.windowSizeMultiple = 2
num_devices = info.get('deviceCount') self.audioInputDeviceIndex = 7
found_device = False self.audioInputChannels = 1
for i in range(0, num_devices):
device_info = pa.get_device_info_by_host_api_device_index(0, i)
if (device_info.get('maxInputChannels')) > 0:
print(f" Input Device id {i} - {device_info.get('name')}")
if i == audioInputDeviceIndex:
found_device = True
if not found_device: self.pa = pyaudio.PyAudio()
print(f"Warning: Audio input device index {audioInputDeviceIndex} not found or has no input channels.")
# Consider exiting or picking a default if necessary
try: logging.info("Available audio input devices:")
audioInputDevice = pa.get_device_info_by_index(audioInputDeviceIndex) info = self.pa.get_host_api_info_by_index(0)
audioInputSampleRate = int(audioInputDevice['defaultSampleRate']) num_devices = info.get('deviceCount')
except Exception as e: found_device = False
print(f"Error getting audio device info for index {audioInputDeviceIndex}: {e}") for i in range(0, num_devices):
pa.terminate() device_info = self.pa.get_device_info_by_host_api_device_index(0, i)
exit() if (device_info.get('maxInputChannels')) > 0:
logging.info(f" Input Device id {i} - {device_info.get('name')}")
if i == self.audioInputDeviceIndex:
found_device = True
# create the aubio tempo detection: if not found_device:
hopSize = bufferSize logging.warning(f"Audio input device index {self.audioInputDeviceIndex} not found or has no input channels.")
winSize = hopSize * windowSizeMultiple
tempoDetection = aubio.tempo(method='default', buf_size=winSize, hop_size=hopSize, samplerate=audioInputSampleRate)
# --- WebSocket Setup --- try:
websocket_url = "ws://192.168.4.1:80/ws" audioInputDevice = self.pa.get_device_info_by_index(self.audioInputDeviceIndex)
ws = None self.audioInputSampleRate = int(audioInputDevice['defaultSampleRate'])
try: except Exception as e:
ws = websocket.create_connection(websocket_url) logging.error(f"Error getting audio device info for index {self.audioInputDeviceIndex}: {e}")
print(f"Successfully connected to WebSocket at {websocket_url}") self.pa.terminate()
except Exception as e: exit()
print(f"Failed to connect to WebSocket: {e}. Data will not be sent over WebSocket.")
# --- End WebSocket Setup ---
# this function gets called by the input stream, as soon as enough samples are collected from the audio input: self.hopSize = self.bufferSize
def readAudioFrames(in_data, frame_count, time_info, status): self.winSize = self.hopSize * self.windowSizeMultiple
global ws # Allow modification of the global ws variable self.tempoDetection = aubio.tempo(method='default', buf_size=self.winSize, hop_size=self.hopSize, samplerate=self.audioInputSampleRate)
signal = np.frombuffer(in_data, dtype=np.float32) self.inputStream = None
self._control_thread = None
beat = tempoDetection(signal) self._connect_to_midi_server()
if beat: self._start_control_server() # Start control server in background
bpm = tempoDetection.get_bpm()
print(f"beat! (running with {bpm:.2f} bpm)") # Use f-string for cleaner formatting, removed extra bells
data_to_send = {
"names": ["1"],
"settings": {
"pattern": "pulse",
"delay": 10,
"colors": ["#00ff00"],
"brightness": 10,
"num_leds": 200,
},
}
if ws: # Only send if the websocket connection is established def reset_tempo_detection(self):
"""Re-initializes the aubio tempo detection object."""
logging.info("[SoundBeatDetector] Resetting tempo detection.")
self.tempoDetection = aubio.tempo(method='default', buf_size=self.winSize, hop_size=self.hopSize, samplerate=self.audioInputSampleRate)
def _control_server_loop(self):
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((SOUND_CONTROL_HOST, SOUND_CONTROL_PORT))
srv.listen(5)
logging.info(f"[SoundBeatDetector - Control] Listening on {SOUND_CONTROL_HOST}:{SOUND_CONTROL_PORT}")
while True:
conn, addr = srv.accept()
with conn:
logging.info(f"[SoundBeatDetector - Control] Connection from {addr}")
try:
data = conn.recv(1024)
if not data:
continue
command = data.decode().strip()
logging.debug(f"[SoundBeatDetector - Control] Received command: {command}")
if command == "RESET_TEMPO":
self.reset_tempo_detection()
response = "OK: Tempo reset\n"
else:
response = "ERROR: Unknown command\n"
conn.sendall(response.encode('utf-8'))
except Exception as e:
logging.error(f"[SoundBeatDetector - Control] Error handling control message: {e}")
except Exception as e:
logging.error(f"[SoundBeatDetector - Control] Server error: {e}")
finally:
try: try:
ws.send(json.dumps(data_to_send)) srv.close()
# print("Sent data over WebSocket") # Optional: for debugging except Exception:
except websocket.WebSocketConnectionClosedException: pass
print("WebSocket connection closed, attempting to reconnect...")
ws = None # Mark as closed, connection will need to be re-established if desired
except Exception as e:
print(f"Error sending over WebSocket: {e}")
return (in_data, pyaudio.paContinue) def _start_control_server(self):
if self._control_thread and self._control_thread.is_alive():
return
self._control_thread = threading.Thread(target=self._control_server_loop, daemon=True)
self._control_thread.start()
def _connect_to_midi_server(self):
if self.tcp_socket:
self.tcp_socket.close()
self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.tcp_socket.settimeout(self.reconnect_delay)
try:
logging.info(f"[SoundBeatDetector] Attempting to connect to MIDI TCP server at {self.tcp_host}:{self.tcp_port}...")
self.tcp_socket.connect((self.tcp_host, self.tcp_port))
self.tcp_socket.setblocking(0)
self.connected_to_midi = True
logging.info(f"[SoundBeatDetector] Successfully connected to MIDI TCP server.")
except (socket.error, socket.timeout) as e:
logging.error(f"[SoundBeatDetector] Failed to connect to MIDI TCP server: {e}")
self.connected_to_midi = False
# create and start the input stream # Removed _handle_control_client and _control_server (replaced by simple threaded server)
try:
inputStream = pa.open(format=pyaudio.paFloat32,
input=True,
channels=audioInputChannels,
input_device_index=audioInputDeviceIndex,
frames_per_buffer=bufferSize,
rate=audioInputSampleRate,
stream_callback=readAudioFrames)
inputStream.start_stream() def readAudioFrames(self, in_data, frame_count, time_info, status):
print("\nAudio stream started. Detecting beats. Press Ctrl+C to stop.") signal = np.frombuffer(in_data, dtype=np.float32)
# Loop to keep the script running, allowing graceful shutdown beat = self.tempoDetection(signal)
while inputStream.is_active(): if beat:
sleep(0.1) # Small delay to prevent busy-waiting bpm = self.tempoDetection.get_bpm()
logging.debug(f"beat! (running with {bpm:.2f} bpm)") # Changed to debug
bpm_message = str(bpm)
except KeyboardInterrupt: if self.connected_to_midi and self.tcp_socket:
print("\nKeyboardInterrupt: Stopping script gracefully.") try:
except Exception as e: message_bytes = (bpm_message + "\n").encode('utf-8')
print(f"An error occurred with the audio stream: {e}") self.tcp_socket.sendall(message_bytes)
finally: logging.debug(f"[SoundBeatDetector] Sent BPM to MIDI TCP server: {bpm_message}") # Changed to debug
# Ensure streams and resources are closed except socket.error as e:
if 'inputStream' in locals() and inputStream.is_active(): logging.error(f"[SoundBeatDetector] Error sending BPM to MIDI TCP server: {e}. Attempting to reconnect...")
inputStream.stop_stream() self.connected_to_midi = False
if 'inputStream' in locals() and not inputStream.is_stopped(): self._connect_to_midi_server()
inputStream.close() elif not self.connected_to_midi:
pa.terminate() logging.warning("[SoundBeatDetector] Not connected to MIDI TCP server, attempting to reconnect...") # Changed to warning
if ws: self._connect_to_midi_server()
print("Closing WebSocket connection.") else:
ws.close() logging.warning("[SoundBeatDetector] TCP socket not initialized, cannot send BPM.") # Changed to warning
print("Script finished.") return (in_data, pyaudio.paContinue)
def start_stream(self):
try:
self.inputStream = self.pa.open(format=pyaudio.paFloat32,
input=True,
channels=self.audioInputChannels,
input_device_index=self.audioInputDeviceIndex,
frames_per_buffer=self.bufferSize,
rate=self.audioInputSampleRate,
stream_callback=self.readAudioFrames)
self.inputStream.start_stream()
logging.info("\nAudio stream started. Detecting beats. Press Ctrl+C to stop.")
while self.inputStream.is_active():
sleep(0.1)
except KeyboardInterrupt:
logging.info("\nKeyboardInterrupt: Stopping script gracefully.")
except Exception as e:
logging.error(f"An error occurred with the audio stream: {e}")
finally:
self.stop_stream()
def stop_stream(self):
if self.inputStream and self.inputStream.is_active():
self.inputStream.stop_stream()
if self.inputStream and not self.inputStream.is_stopped():
self.inputStream.close()
self.pa.terminate()
if self.tcp_socket and self.connected_to_midi:
logging.info("[SoundBeatDetector] Closing TCP socket.")
self.tcp_socket.close()
self.connected_to_midi = False
logging.info("SoundBeatDetector stopped.")
# Removed async def run(self)
if __name__ == "__main__":
# TCP Server Configuration (should match midi.py)
MIDI_TCP_HOST = "127.0.0.1"
MIDI_TCP_PORT = 65432
sound_detector = SoundBeatDetector(MIDI_TCP_HOST, MIDI_TCP_PORT)
logging.info("Starting SoundBeatDetector...")
try:
sound_detector.start_stream()
except KeyboardInterrupt:
logging.info("\nProgram interrupted by user.")
except Exception as e:
logging.error(f"An error occurred during main execution: {e}")