Listen to your servers scream in agony through a MIDI synthesizer
Initial checkin
+1285
+13
.gitignore
+13
.gitignore
+21
LICENSE
+21
LICENSE
···
1
+
MIT License
2
+
3
+
Copyright (c) 2025
4
+
5
+
Permission is hereby granted, free of charge, to any person obtaining a copy
6
+
of this software and associated documentation files (the "Software"), to deal
7
+
in the Software without restriction, including without limitation the rights
8
+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9
+
copies of the Software, and to permit persons to whom the Software is
10
+
furnished to do so, subject to the following conditions:
11
+
12
+
The above copyright notice and this permission notice shall be included in all
13
+
copies or substantial portions of the Software.
14
+
15
+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16
+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17
+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18
+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19
+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20
+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21
+
SOFTWARE.
+150
README.md
+150
README.md
···
1
+
# snmp-midi
2
+
3
+
Polls server metrics via SNMP and plays them as MIDI notes. Each stat maps to an instrument on its own channel — pitch tracks the value, velocity tracks intensity. The result is an ambient, real-time sonic portrait of a running server.
4
+
5
+
## Features
6
+
7
+
- Maps CPU, memory, disk, network, and process stats to MIDI instruments
8
+
- Per-core CPU load and multi-timescale load averages (1min / 5min / 15min)
9
+
- Configurable scales, octaves, note durations, and velocity ranges
10
+
- Cascade stagger: voices roll in sequentially each poll cycle
11
+
- Optional threshold triggers for alert notes
12
+
- Works with hardware synths, virtual ports, or any MIDI-capable DAW
13
+
- Roland GS reset on startup for hardware synth initialization
14
+
15
+
## Requirements
16
+
17
+
- Python 3.8+
18
+
- SNMP-enabled server (`snmpd` with UCD-SNMP-MIB and HOST-RESOURCES-MIB)
19
+
- MIDI output: hardware synth, software synth (FluidSynth), or DAW
20
+
21
+
## Installation
22
+
23
+
```bash
24
+
python3 -m venv venv
25
+
source venv/bin/activate
26
+
pip install -r requirements.txt
27
+
```
28
+
29
+
## Quick Start
30
+
31
+
```bash
32
+
# Copy example config and edit with your server details
33
+
cp config.example.yaml config.yaml
34
+
$EDITOR config.yaml
35
+
36
+
# List available MIDI ports
37
+
./main.py -l
38
+
39
+
# Run (port number from -l output)
40
+
./main.py -P 0
41
+
```
42
+
43
+
## Usage
44
+
45
+
```
46
+
./main.py [options]
47
+
48
+
-c FILE Config file (default: config.yaml)
49
+
-g Write default config to FILE and exit
50
+
-l List MIDI ports and exit
51
+
-H HOST SNMP host (overrides config)
52
+
-p PORT SNMP port (overrides config)
53
+
-C STR SNMP community string (overrides config)
54
+
-P NUM MIDI port by index number
55
+
-m NAME MIDI port by name
56
+
-v Use virtual MIDI port
57
+
-i SECS Poll interval in seconds (overrides config)
58
+
```
59
+
60
+
## Configuration
61
+
62
+
Copy `config.example.yaml` to `config.yaml` and edit. The `config.yaml` file is gitignored so credentials stay local.
63
+
64
+
### Available stat types
65
+
66
+
| `stat_name` | `stat_key` | Description |
67
+
|--------------------|---------------------------------------------|------------------------------------|
68
+
| `cpu_load` | — | 1-minute load average |
69
+
| `cpu_load_5min` | — | 5-minute load average |
70
+
| `cpu_load_15min` | — | 15-minute load average |
71
+
| `per_core_cpu_load`| `core0`, `core1`, … `coreN` | Per-core CPU % (hrProcessorLoad) |
72
+
| `memory_usage` | `percent`, `total`, `available`, `cached` | Memory stats (UCD-SNMP-MIB) |
73
+
| `disk_usage` | `percent`, `size`, `used` | First mounted filesystem |
74
+
| `network_stats` | `in_bytes`, `out_bytes` | Cumulative octets (IF-MIB) |
75
+
| `process_count` | — | hrSystemProcesses |
76
+
77
+
### Scales
78
+
79
+
`major` · `minor` · `pentatonic_major` · `pentatonic_minor` · `blues` · `dorian` · `mixolydian` · `chromatic`
80
+
81
+
### Mapping fields
82
+
83
+
```yaml
84
+
mappings:
85
+
- stat_name: cpu_load # stat type (see table above)
86
+
stat_key: null # sub-key within stat, if applicable
87
+
channel: 0 # MIDI channel 0–15
88
+
program: 73 # GM program number 0–127
89
+
scale: pentatonic_major # musical scale
90
+
root: C # root note
91
+
octave: 5 # base octave
92
+
min_value: 0 # value mapped to lowest scale note
93
+
max_value: 10 # value mapped to highest scale note
94
+
note_duration: 1.0 # seconds (must be < poll_interval)
95
+
cc_number: null # also send value as CC, or null
96
+
enabled: true
97
+
```
98
+
99
+
### Threshold triggers
100
+
101
+
```yaml
102
+
triggers:
103
+
- stat_name: cpu_load
104
+
threshold: 8.0
105
+
above: true
106
+
note: 72
107
+
velocity: 127
108
+
channel: 0
109
+
duration: 0.5
110
+
```
111
+
112
+
### Timing
113
+
114
+
`note_duration` must be less than `poll_interval` or note-off threads from earlier polls will cancel newer notes before slow-attack instruments sound. Use onboard reverb to blur the gaps between notes.
115
+
116
+
## Setting Up snmpd
117
+
118
+
```bash
119
+
sudo apt-get install snmpd snmp-mibs-downloader
120
+
```
121
+
122
+
`/etc/snmp/snmpd.conf` minimal additions:
123
+
```
124
+
rocommunity public 127.0.0.1
125
+
rocommunity public 192.168.0.0/16
126
+
```
127
+
128
+
```bash
129
+
sudo systemctl restart snmpd
130
+
# Test
131
+
snmpwalk -v2c -c public localhost 1.3.6.1.2.1.1.1.0
132
+
```
133
+
134
+
## Connecting a Synthesizer
135
+
136
+
**Hardware synth:** connect via USB-MIDI or DIN, then `./main.py -l` to find the port index.
137
+
138
+
**FluidSynth:**
139
+
```bash
140
+
sudo apt-get install fluidsynth fluid-soundfont-gm
141
+
fluidsynth -a alsa -m alsa_seq /usr/share/sounds/sf2/FluidR3_GM.sf2 &
142
+
./main.py -l
143
+
./main.py -P <port number>
144
+
```
145
+
146
+
**DAW:** create a virtual MIDI port in your DAW, then `./main.py -m "port name"`.
147
+
148
+
## License
149
+
150
+
[MIT](LICENSE)
+119
config.example.yaml
+119
config.example.yaml
···
1
+
snmp:
2
+
host: 192.168.1.100 # your SNMP target
3
+
port: 161
4
+
community: public
5
+
version: 2
6
+
timeout: 5
7
+
retries: 3
8
+
9
+
midi:
10
+
port_name: null # null = auto-detect first port
11
+
virtual: false
12
+
bpm: 120
13
+
poll_interval: 1.1
14
+
15
+
# All voices use C pentatonic major (C D E G A) — every combination consonant.
16
+
#
17
+
# Temporal layers (slow → fast):
18
+
# cpu_load_15min → strings pad, changes over 15+ min
19
+
# cpu_load_5min → choir, changes over 5+ min
20
+
# cpu_load → flute, 1-min average, responsive melody
21
+
#
22
+
# Instantaneous per-core CPU:
23
+
# core0 → vibraphone core1 → marimba
24
+
# core2 → celesta core3 → glockenspiel
25
+
#
26
+
# Under load: cores jump immediately, load averages follow over minutes.
27
+
# Adjust max_value for your server's typical load range.
28
+
29
+
mappings:
30
+
31
+
- stat_name: cpu_load_15min
32
+
channel: 0
33
+
program: 48 # String Ensemble 1
34
+
scale: pentatonic_major
35
+
root: C
36
+
octave: 2
37
+
min_value: 0
38
+
max_value: 8
39
+
note_duration: 1.0
40
+
cc_number: null
41
+
enabled: true
42
+
43
+
- stat_name: cpu_load_5min
44
+
channel: 1
45
+
program: 52 # Choir Aahs
46
+
scale: pentatonic_major
47
+
root: C
48
+
octave: 3
49
+
min_value: 0
50
+
max_value: 8
51
+
note_duration: 1.0
52
+
cc_number: null
53
+
enabled: true
54
+
55
+
- stat_name: cpu_load
56
+
channel: 2
57
+
program: 73 # Flute
58
+
scale: pentatonic_major
59
+
root: C
60
+
octave: 5
61
+
min_value: 0
62
+
max_value: 10
63
+
note_duration: 1.0
64
+
cc_number: null
65
+
enabled: true
66
+
67
+
- stat_name: per_core_cpu_load
68
+
stat_key: core0
69
+
channel: 3
70
+
program: 11 # Vibraphone
71
+
scale: pentatonic_major
72
+
root: C
73
+
octave: 3
74
+
min_value: 0
75
+
max_value: 30
76
+
note_duration: 1.0
77
+
cc_number: null
78
+
enabled: true
79
+
80
+
- stat_name: per_core_cpu_load
81
+
stat_key: core1
82
+
channel: 4
83
+
program: 12 # Marimba
84
+
scale: pentatonic_major
85
+
root: C
86
+
octave: 4
87
+
min_value: 0
88
+
max_value: 30
89
+
note_duration: 1.0
90
+
cc_number: null
91
+
enabled: true
92
+
93
+
- stat_name: per_core_cpu_load
94
+
stat_key: core2
95
+
channel: 5
96
+
program: 8 # Celesta
97
+
scale: pentatonic_major
98
+
root: C
99
+
octave: 4
100
+
min_value: 0
101
+
max_value: 30
102
+
note_duration: 1.0
103
+
cc_number: null
104
+
enabled: true
105
+
106
+
- stat_name: per_core_cpu_load
107
+
stat_key: core3
108
+
channel: 6
109
+
program: 9 # Glockenspiel
110
+
scale: pentatonic_major
111
+
root: C
112
+
octave: 5
113
+
min_value: 0
114
+
max_value: 30
115
+
note_duration: 1.0
116
+
cc_number: null
117
+
enabled: true
118
+
119
+
triggers: []
+229
config.py
+229
config.py
···
1
+
# SPDX-License-Identifier: MIT
2
+
3
+
import yaml
4
+
import os
5
+
from typing import Dict, List, Any, Optional
6
+
from dataclasses import dataclass, field
7
+
8
+
9
+
@dataclass
10
+
class SNMPConfig:
11
+
host: str = 'localhost'
12
+
port: int = 161
13
+
community: str = 'public'
14
+
version: int = 2
15
+
timeout: int = 5
16
+
retries: int = 3
17
+
18
+
19
+
@dataclass
20
+
class MIDIConfig:
21
+
port_name: Optional[str] = None
22
+
virtual: bool = True
23
+
bpm: int = 120
24
+
25
+
26
+
@dataclass
27
+
class StatMapping:
28
+
stat_name: str
29
+
stat_key: Optional[str] = None
30
+
channel: int = 0
31
+
program: int = 0
32
+
scale: str = 'major'
33
+
root: str = 'C'
34
+
octave: int = 4
35
+
min_value: float = 0
36
+
max_value: float = 100
37
+
note_duration: float = 0.5
38
+
cc_number: Optional[int] = None
39
+
enabled: bool = True
40
+
41
+
42
+
@dataclass
43
+
class TriggerMapping:
44
+
stat_name: str
45
+
stat_key: Optional[str] = None
46
+
threshold: float = 80
47
+
above: bool = True
48
+
note: int = 60
49
+
velocity: int = 127
50
+
channel: int = 0
51
+
duration: float = 1.0
52
+
53
+
54
+
@dataclass
55
+
class Config:
56
+
snmp: SNMPConfig = field(default_factory=SNMPConfig)
57
+
midi: MIDIConfig = field(default_factory=MIDIConfig)
58
+
stat_mappings: List[StatMapping] = field(default_factory=list)
59
+
triggers: List[TriggerMapping] = field(default_factory=list)
60
+
poll_interval: float = 1.0
61
+
62
+
@classmethod
63
+
def from_yaml(cls, filepath: str) -> 'Config':
64
+
with open(filepath, 'r') as f:
65
+
data = yaml.safe_load(f)
66
+
return cls.from_dict(data)
67
+
68
+
@classmethod
69
+
def from_dict(cls, data: Dict[str, Any]) -> 'Config':
70
+
config = cls()
71
+
72
+
if 'snmp' in data:
73
+
d = data['snmp']
74
+
config.snmp = SNMPConfig(
75
+
host=d.get('host', 'localhost'),
76
+
port=d.get('port', 161),
77
+
community=d.get('community', 'public'),
78
+
version=d.get('version', 2),
79
+
timeout=d.get('timeout', 5),
80
+
retries=d.get('retries', 3),
81
+
)
82
+
83
+
if 'midi' in data:
84
+
d = data['midi']
85
+
config.midi = MIDIConfig(
86
+
port_name=d.get('port_name'),
87
+
virtual=d.get('virtual', True),
88
+
bpm=d.get('bpm', 120),
89
+
)
90
+
91
+
config.poll_interval = data.get('poll_interval', 1.0)
92
+
93
+
for m in data.get('mappings', []):
94
+
config.stat_mappings.append(StatMapping(
95
+
stat_name=m['stat_name'],
96
+
stat_key=m.get('stat_key'),
97
+
channel=m.get('channel', 0),
98
+
program=m.get('program', 0),
99
+
scale=m.get('scale', 'major'),
100
+
root=m.get('root', 'C'),
101
+
octave=m.get('octave', 4),
102
+
min_value=m.get('min_value', 0),
103
+
max_value=m.get('max_value', 100),
104
+
note_duration=m.get('note_duration', 0.5),
105
+
cc_number=m.get('cc_number'),
106
+
enabled=m.get('enabled', True),
107
+
))
108
+
109
+
for t in data.get('triggers', []):
110
+
config.triggers.append(TriggerMapping(
111
+
stat_name=t['stat_name'],
112
+
stat_key=t.get('stat_key'),
113
+
threshold=t.get('threshold', 80),
114
+
above=t.get('above', True),
115
+
note=t.get('note', 60),
116
+
velocity=t.get('velocity', 127),
117
+
channel=t.get('channel', 0),
118
+
duration=t.get('duration', 1.0),
119
+
))
120
+
121
+
return config
122
+
123
+
def to_yaml(self, filepath: str):
124
+
data = {
125
+
'snmp': {
126
+
'host': self.snmp.host, 'port': self.snmp.port,
127
+
'community': self.snmp.community, 'version': self.snmp.version,
128
+
'timeout': self.snmp.timeout, 'retries': self.snmp.retries,
129
+
},
130
+
'midi': {
131
+
'port_name': self.midi.port_name,
132
+
'virtual': self.midi.virtual,
133
+
'bpm': self.midi.bpm,
134
+
},
135
+
'poll_interval': self.poll_interval,
136
+
'mappings': [
137
+
{
138
+
'stat_name': m.stat_name, 'stat_key': m.stat_key,
139
+
'channel': m.channel, 'program': m.program,
140
+
'scale': m.scale, 'root': m.root, 'octave': m.octave,
141
+
'min_value': m.min_value, 'max_value': m.max_value,
142
+
'note_duration': m.note_duration, 'cc_number': m.cc_number,
143
+
'enabled': m.enabled,
144
+
}
145
+
for m in self.stat_mappings
146
+
],
147
+
'triggers': [
148
+
{
149
+
'stat_name': t.stat_name, 'stat_key': t.stat_key,
150
+
'threshold': t.threshold, 'above': t.above,
151
+
'note': t.note, 'velocity': t.velocity,
152
+
'channel': t.channel, 'duration': t.duration,
153
+
}
154
+
for t in self.triggers
155
+
],
156
+
}
157
+
with open(filepath, 'w') as f:
158
+
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
159
+
160
+
def to_dict(self) -> Dict[str, Any]:
161
+
return {
162
+
'snmp': {
163
+
'host': self.snmp.host, 'port': self.snmp.port,
164
+
'community': self.snmp.community, 'version': self.snmp.version,
165
+
'timeout': self.snmp.timeout, 'retries': self.snmp.retries,
166
+
},
167
+
'midi': {
168
+
'port_name': self.midi.port_name,
169
+
'virtual': self.midi.virtual,
170
+
'bpm': self.midi.bpm,
171
+
},
172
+
'poll_interval': self.poll_interval,
173
+
'mappings': [
174
+
{
175
+
'stat_name': m.stat_name, 'stat_key': m.stat_key,
176
+
'channel': m.channel, 'program': m.program,
177
+
'scale': m.scale, 'root': m.root, 'octave': m.octave,
178
+
'min_value': m.min_value, 'max_value': m.max_value,
179
+
'note_duration': m.note_duration, 'cc_number': m.cc_number,
180
+
'enabled': m.enabled,
181
+
}
182
+
for m in self.stat_mappings
183
+
],
184
+
'triggers': [
185
+
{
186
+
'stat_name': t.stat_name, 'stat_key': t.stat_key,
187
+
'threshold': t.threshold, 'above': t.above,
188
+
'note': t.note, 'velocity': t.velocity,
189
+
'channel': t.channel, 'duration': t.duration,
190
+
}
191
+
for t in self.triggers
192
+
],
193
+
}
194
+
195
+
196
+
def create_default_config() -> Config:
197
+
config = Config()
198
+
config.stat_mappings.append(StatMapping(
199
+
stat_name='cpu_load', channel=0, program=80,
200
+
scale='pentatonic_minor', root='C', octave=5,
201
+
min_value=0, max_value=4.0, note_duration=0.3, cc_number=1,
202
+
))
203
+
config.stat_mappings.append(StatMapping(
204
+
stat_name='memory_usage', stat_key='percent', channel=1, program=89,
205
+
scale='major', root='C', octave=3,
206
+
min_value=0, max_value=100, note_duration=2.0, cc_number=11,
207
+
))
208
+
config.stat_mappings.append(StatMapping(
209
+
stat_name='disk_usage', stat_key='percent', channel=2, program=38,
210
+
scale='minor', root='E', octave=2,
211
+
min_value=0, max_value=100, note_duration=0.5,
212
+
))
213
+
config.triggers.append(TriggerMapping(
214
+
stat_name='cpu_load', threshold=3.0, above=True,
215
+
note=72, velocity=127, channel=0, duration=0.5,
216
+
))
217
+
config.triggers.append(TriggerMapping(
218
+
stat_name='memory_usage', stat_key='percent', threshold=90, above=True,
219
+
note=60, velocity=127, channel=1, duration=1.0,
220
+
))
221
+
return config
222
+
223
+
224
+
def load_config(filepath: str) -> Config:
225
+
if os.path.exists(filepath):
226
+
return Config.from_yaml(filepath)
227
+
config = create_default_config()
228
+
config.to_yaml(filepath)
229
+
return config
+347
main.py
+347
main.py
···
1
+
#!/usr/bin/env python3
2
+
# SPDX-License-Identifier: MIT
3
+
4
+
import asyncio
5
+
import argparse
6
+
import sys
7
+
from typing import Dict, Any, Optional
8
+
from datetime import datetime
9
+
10
+
from snmp_client import SNMPClient
11
+
from midi_output import MIDIOutput, MIDIInstrument
12
+
from config import Config, load_config, create_default_config
13
+
14
+
15
+
class SNMPSynth:
16
+
def __init__(self, config: Config):
17
+
self.config = config
18
+
self.snmp_client: Optional[SNMPClient] = None
19
+
self.midi_output: Optional[MIDIOutput] = None
20
+
self.instruments: Dict[str, MIDIInstrument] = {}
21
+
self.running = False
22
+
self.last_values: Dict[str, Any] = {}
23
+
self._ticker_pos = 0
24
+
25
+
async def initialize(self):
26
+
print(f"Connecting to SNMP host: {self.config.snmp.host}")
27
+
self.snmp_client = SNMPClient(
28
+
host=self.config.snmp.host,
29
+
port=self.config.snmp.port,
30
+
community=self.config.snmp.community,
31
+
version=self.config.snmp.version,
32
+
timeout=self.config.snmp.timeout,
33
+
retries=self.config.snmp.retries,
34
+
)
35
+
36
+
print("Opening MIDI output...")
37
+
self.midi_output = MIDIOutput(
38
+
port_name=self.config.midi.port_name,
39
+
virtual=self.config.midi.virtual,
40
+
)
41
+
self.midi_output.open()
42
+
43
+
print("Sending GS Reset...")
44
+
self.midi_output.send_gs_reset()
45
+
await asyncio.sleep(0.5)
46
+
47
+
for mapping in self.config.stat_mappings:
48
+
if not mapping.enabled:
49
+
continue
50
+
instrument = MIDIInstrument(
51
+
output=self.midi_output,
52
+
channel=mapping.channel,
53
+
program=mapping.program,
54
+
scale=mapping.scale,
55
+
root=mapping.root,
56
+
octave=mapping.octave,
57
+
)
58
+
instrument.setup()
59
+
self.midi_output.send_cc(7, 100, mapping.channel)
60
+
self.midi_output.send_cc(91, 80, mapping.channel)
61
+
self.midi_output.send_cc(93, 20, mapping.channel)
62
+
key = f"{mapping.stat_name}:{mapping.stat_key or 'default'}"
63
+
self.instruments[key] = instrument
64
+
65
+
print(f"Initialized {len(self.instruments)} instruments")
66
+
67
+
async def get_stat_value(self, mapping) -> Optional[float]:
68
+
try:
69
+
if mapping.stat_name == 'cpu_load':
70
+
return await self.snmp_client.get_cpu_load()
71
+
72
+
elif mapping.stat_name == 'cpu_load_5min':
73
+
return await self.snmp_client.get_cpu_load_5min()
74
+
75
+
elif mapping.stat_name == 'cpu_load_15min':
76
+
return await self.snmp_client.get_cpu_load_15min()
77
+
78
+
elif mapping.stat_name == 'cpu_core_stats':
79
+
stats = await self.snmp_client.get_cpu_core_stats()
80
+
if mapping.stat_key and mapping.stat_key in stats:
81
+
return float(stats[mapping.stat_key])
82
+
83
+
elif mapping.stat_name == 'per_core_cpu_load':
84
+
per_core = await self.snmp_client.get_per_core_cpu_load()
85
+
if mapping.stat_key and mapping.stat_key in per_core:
86
+
return float(per_core[mapping.stat_key])
87
+
88
+
elif mapping.stat_name == 'memory_usage':
89
+
mem = await self.snmp_client.get_memory_usage()
90
+
total = mem.get('total', 0)
91
+
if mapping.stat_key == 'percent' or mapping.stat_key is None:
92
+
available = mem.get('available', 0)
93
+
return ((total - available) / total * 100) if total > 0 else None
94
+
elif mapping.stat_key:
95
+
return float(mem.get(mapping.stat_key, 0))
96
+
97
+
elif mapping.stat_name == 'disk_usage':
98
+
disks = await self.snmp_client.get_disk_usage()
99
+
if disks:
100
+
key = mapping.stat_key or 'percent'
101
+
return float(disks[0].get(key, 0))
102
+
103
+
elif mapping.stat_name == 'network_stats':
104
+
interfaces = await self.snmp_client.get_network_stats()
105
+
if interfaces:
106
+
if mapping.stat_key == 'in_bytes':
107
+
return sum(i.get('in_bytes', 0) for i in interfaces)
108
+
elif mapping.stat_key == 'out_bytes':
109
+
return sum(i.get('out_bytes', 0) for i in interfaces)
110
+
return sum(i.get('in_bytes', 0) + i.get('out_bytes', 0) for i in interfaces)
111
+
112
+
elif mapping.stat_name == 'process_count':
113
+
return float(await self.snmp_client.get_process_count())
114
+
115
+
except Exception as e:
116
+
print(f"\nError getting {mapping.stat_name}: {e}")
117
+
118
+
return None
119
+
120
+
async def check_triggers(self, stats: Dict[str, Any]):
121
+
for trigger in self.config.triggers:
122
+
value = None
123
+
if trigger.stat_name == 'cpu_load':
124
+
value = stats.get('cpu_load')
125
+
elif trigger.stat_name == 'memory_usage':
126
+
mem = stats.get('memory_usage', {})
127
+
total = mem.get('total', 0)
128
+
if trigger.stat_key == 'percent' and total > 0:
129
+
value = (total - mem.get('available', 0)) / total * 100
130
+
elif trigger.stat_key:
131
+
value = mem.get(trigger.stat_key)
132
+
elif trigger.stat_name == 'disk_usage':
133
+
disks = stats.get('disk_usage', [])
134
+
if disks and trigger.stat_key:
135
+
value = disks[0].get(trigger.stat_key)
136
+
elif trigger.stat_name == 'process_count':
137
+
value = stats.get('process_count')
138
+
139
+
if value is None:
140
+
continue
141
+
142
+
fired = (trigger.above and value > trigger.threshold) or \
143
+
(not trigger.above and value < trigger.threshold)
144
+
if fired:
145
+
print(f"Trigger: {trigger.stat_name} {value:.1f} "
146
+
f"{'>' if trigger.above else '<'} {trigger.threshold}")
147
+
self.midi_output.play_note(
148
+
trigger.note, trigger.duration, trigger.velocity, trigger.channel
149
+
)
150
+
151
+
async def poll_once(self):
152
+
stats = {}
153
+
154
+
try:
155
+
stats['cpu_load'] = await self.snmp_client.get_cpu_load()
156
+
if stats['cpu_load'] is None:
157
+
alt = await self.snmp_client.get_cpu_load_alternative()
158
+
if alt is not None:
159
+
stats['cpu_load'] = alt
160
+
except Exception as e:
161
+
print(f"\n[ERROR] cpu_load: {e}")
162
+
stats['cpu_load'] = None
163
+
164
+
try:
165
+
stats['memory_usage'] = await self.snmp_client.get_memory_usage()
166
+
except Exception as e:
167
+
print(f"\n[ERROR] memory_usage: {e}")
168
+
stats['memory_usage'] = {}
169
+
170
+
try:
171
+
stats['disk_usage'] = await self.snmp_client.get_disk_usage()
172
+
except Exception as e:
173
+
print(f"\n[ERROR] disk_usage: {e}")
174
+
stats['disk_usage'] = []
175
+
176
+
try:
177
+
stats['network_stats'] = await self.snmp_client.get_network_stats()
178
+
except Exception as e:
179
+
print(f"\n[ERROR] network_stats: {e}")
180
+
stats['network_stats'] = []
181
+
182
+
try:
183
+
stats['process_count'] = await self.snmp_client.get_process_count()
184
+
except Exception as e:
185
+
stats['process_count'] = 0
186
+
187
+
try:
188
+
stats['cpu_load_5min'] = await self.snmp_client.get_cpu_load_5min()
189
+
except Exception:
190
+
stats['cpu_load_5min'] = None
191
+
192
+
try:
193
+
stats['cpu_load_15min'] = await self.snmp_client.get_cpu_load_15min()
194
+
except Exception:
195
+
stats['cpu_load_15min'] = None
196
+
197
+
enabled_count = sum(1 for m in self.config.stat_mappings if m.enabled)
198
+
stagger_delay = min(0.10, self.config.poll_interval / (enabled_count * 2)) if enabled_count else 0
199
+
200
+
for i, mapping in enumerate(m for m in self.config.stat_mappings if m.enabled):
201
+
value = await self.get_stat_value(mapping)
202
+
if value is None:
203
+
continue
204
+
205
+
key = f"{mapping.stat_name}:{mapping.stat_key or 'default'}"
206
+
instrument = self.instruments.get(key)
207
+
if not instrument:
208
+
continue
209
+
210
+
if i > 0:
211
+
await asyncio.sleep(stagger_delay)
212
+
213
+
instrument.play_value(value, mapping.min_value, mapping.max_value, mapping.note_duration)
214
+
215
+
if mapping.cc_number is not None:
216
+
instrument.send_cc_value(mapping.cc_number, value, mapping.min_value, mapping.max_value)
217
+
218
+
self.last_values[key] = value
219
+
220
+
await self.check_triggers(stats)
221
+
self.display_values()
222
+
# self.update_display(stats)
223
+
224
+
def update_display(self, stats: Dict[str, Any]):
225
+
if not self.midi_output:
226
+
return
227
+
228
+
parts = []
229
+
for label, key in [('L1', 'cpu_load'), ('L5', 'cpu_load_5min'), ('L15', 'cpu_load_15min')]:
230
+
v = stats.get(key)
231
+
if v is not None:
232
+
parts.append(f"{label}:{v:.1f}")
233
+
for i in range(8):
234
+
v = self.last_values.get(f"per_core_cpu_load:core{i}")
235
+
if v is not None:
236
+
parts.append(f"C{i}:{int(v)}%")
237
+
mem = self.last_values.get("memory_usage:percent")
238
+
if mem is not None:
239
+
parts.append(f"MEM:{mem:.0f}%")
240
+
procs = stats.get('process_count')
241
+
if procs is not None:
242
+
parts.append(f"P:{int(procs)}")
243
+
244
+
if not parts:
245
+
return
246
+
ticker = " | ".join(parts) + " | "
247
+
pos = self._ticker_pos % len(ticker)
248
+
window = (ticker + ticker)[pos:pos + 16]
249
+
self._ticker_pos = (pos + 3) % len(ticker)
250
+
self.midi_output.send_display_text(window)
251
+
252
+
def display_values(self):
253
+
ts = datetime.now().strftime("%H:%M:%S")
254
+
vals = " | ".join(f"{k.split(':')[0]}: {v:.1f}" for k, v in self.last_values.items())
255
+
print(f"\r[{ts}] {vals}", end="", flush=True)
256
+
257
+
async def run(self):
258
+
await self.initialize()
259
+
self.running = True
260
+
print(f"\nSNMP-MIDI running (poll interval: {self.config.poll_interval}s)")
261
+
print("Press Ctrl+C to stop\n")
262
+
try:
263
+
while self.running:
264
+
await self.poll_once()
265
+
await asyncio.sleep(self.config.poll_interval)
266
+
except KeyboardInterrupt:
267
+
print("\n\nStopping...")
268
+
finally:
269
+
await self.shutdown()
270
+
271
+
async def shutdown(self):
272
+
self.running = False
273
+
if self.midi_output:
274
+
try:
275
+
self.midi_output.all_notes_off()
276
+
self.midi_output.close()
277
+
except Exception as e:
278
+
print(f"Warning: Error closing MIDI: {e}")
279
+
print("SNMP-MIDI stopped")
280
+
281
+
282
+
def list_midi_ports():
283
+
print("Available MIDI output ports:")
284
+
ports = MIDIOutput.list_ports()
285
+
if ports:
286
+
for i, port in enumerate(ports):
287
+
print(f" {i}: {port}")
288
+
else:
289
+
print(" No ports found (virtual port will be created)")
290
+
291
+
292
+
async def main():
293
+
parser = argparse.ArgumentParser(description="SNMP-MIDI: Convert server stats to MIDI")
294
+
parser.add_argument('-c', '--config', default='config.yaml')
295
+
parser.add_argument('-g', '--generate-config', action='store_true')
296
+
parser.add_argument('-l', '--list-ports', action='store_true')
297
+
parser.add_argument('-H', '--host')
298
+
parser.add_argument('-p', '--port', type=int)
299
+
parser.add_argument('-C', '--community')
300
+
parser.add_argument('-m', '--midi-port')
301
+
parser.add_argument('-P', '--port-number', type=int)
302
+
parser.add_argument('-v', '--virtual', action='store_true')
303
+
parser.add_argument('-i', '--interval', type=float)
304
+
args = parser.parse_args()
305
+
306
+
if args.list_ports:
307
+
list_midi_ports()
308
+
return
309
+
310
+
if args.generate_config:
311
+
config = create_default_config()
312
+
config.to_yaml(args.config)
313
+
print(f"Default configuration written to: {args.config}")
314
+
return
315
+
316
+
config = load_config(args.config)
317
+
318
+
if args.host:
319
+
config.snmp.host = args.host
320
+
if args.port:
321
+
config.snmp.port = args.port
322
+
if args.community:
323
+
config.snmp.community = args.community
324
+
if args.port_number is not None:
325
+
ports = MIDIOutput.list_ports()
326
+
if 0 <= args.port_number < len(ports):
327
+
config.midi.port_name = ports[args.port_number]
328
+
config.midi.virtual = False
329
+
print(f"Using MIDI port {args.port_number}: {config.midi.port_name}")
330
+
else:
331
+
print(f"Error: Port {args.port_number} out of range (0-{len(ports)-1})")
332
+
for i, p in enumerate(ports):
333
+
print(f" {i}: {p}")
334
+
sys.exit(1)
335
+
if args.midi_port:
336
+
config.midi.port_name = args.midi_port
337
+
config.midi.virtual = False
338
+
if args.virtual:
339
+
config.midi.virtual = True
340
+
if args.interval:
341
+
config.poll_interval = args.interval
342
+
343
+
await SNMPSynth(config).run()
344
+
345
+
346
+
if __name__ == '__main__':
347
+
asyncio.run(main())
+168
midi_output.py
+168
midi_output.py
···
1
+
# SPDX-License-Identifier: MIT
2
+
3
+
import mido
4
+
import mido.backends.rtmidi
5
+
import time
6
+
import threading
7
+
from typing import List, Dict, Optional
8
+
9
+
10
+
class MIDIOutput:
11
+
NOTE_MAP = {
12
+
'C': 0, 'C#': 1, 'Db': 1, 'D': 2, 'D#': 3, 'Eb': 3,
13
+
'E': 4, 'F': 5, 'F#': 6, 'Gb': 6, 'G': 7, 'G#': 8,
14
+
'Ab': 8, 'A': 9, 'A#': 10, 'Bb': 10, 'B': 11,
15
+
}
16
+
17
+
SCALES = {
18
+
'major': [0, 2, 4, 5, 7, 9, 11],
19
+
'minor': [0, 2, 3, 5, 7, 8, 10],
20
+
'pentatonic_major': [0, 2, 4, 7, 9],
21
+
'pentatonic_minor': [0, 3, 5, 7, 10],
22
+
'blues': [0, 3, 5, 6, 7, 10],
23
+
'dorian': [0, 2, 3, 5, 7, 9, 10],
24
+
'mixolydian': [0, 2, 4, 5, 7, 9, 10],
25
+
'chromatic': list(range(12)),
26
+
}
27
+
28
+
def __init__(self, port_name: Optional[str] = None, virtual: bool = False):
29
+
self.port = None
30
+
self.port_name = port_name
31
+
self.virtual = virtual
32
+
self.active_notes: Dict[tuple, float] = {}
33
+
self._lock = threading.Lock()
34
+
35
+
def open(self):
36
+
if self.port is not None:
37
+
return
38
+
try:
39
+
if self.virtual:
40
+
self.port = mido.open_output(self.port_name or 'snmp-midi', virtual=True)
41
+
elif self.port_name:
42
+
self.port = mido.open_output(self.port_name)
43
+
else:
44
+
ports = mido.get_output_names()
45
+
name = ports[0] if ports else 'snmp-midi'
46
+
self.port = mido.open_output(name, virtual=not bool(ports))
47
+
self.port_name = name
48
+
print(f"MIDI output opened: {self.port_name}")
49
+
except Exception as e:
50
+
raise Exception(f"Failed to open MIDI output: {e}")
51
+
52
+
def close(self):
53
+
if self.port:
54
+
for ch in range(16):
55
+
self.port.send(mido.Message('control_change', channel=ch, control=123, value=0))
56
+
self.port.close()
57
+
self.port = None
58
+
print("MIDI output closed")
59
+
60
+
def send_gs_reset(self):
61
+
if self.port:
62
+
self.port.send(mido.Message('sysex', data=[0x41, 0x10, 0x42, 0x12, 0x40, 0x00, 0x7F, 0x00, 0x41]))
63
+
64
+
def send_display_text(self, text: str, device_id: int = 0x10):
65
+
if not self.port:
66
+
return
67
+
padded = text[:16].ljust(16)
68
+
data = [ord(c) & 0x7F for c in padded]
69
+
addr = [0x10, 0x00, 0x00]
70
+
checksum = (0x80 - sum(addr + data) % 0x80) % 0x80
71
+
self.port.send(mido.Message('sysex', data=[0x41, device_id, 0x45, 0x12] + addr + data + [checksum]))
72
+
73
+
def note_to_midi(self, note: str, octave: int = 4) -> int:
74
+
return (octave + 1) * 12 + self.NOTE_MAP.get(note.upper(), 0)
75
+
76
+
def get_scale_notes(self, root: str, scale: str, octave: int = 4, num_notes: int = 7) -> List[int]:
77
+
root_midi = self.note_to_midi(root, octave)
78
+
intervals = self.SCALES.get(scale, self.SCALES['major'])
79
+
notes = []
80
+
for i in range(num_notes):
81
+
note = root_midi + intervals[i % len(intervals)] + (i // len(intervals)) * 12
82
+
if 0 <= note <= 127:
83
+
notes.append(note)
84
+
return notes
85
+
86
+
def value_to_note(self, value: float, min_val: float, max_val: float, scale_notes: List[int]) -> int:
87
+
if max_val == min_val:
88
+
return scale_notes[len(scale_notes) // 2]
89
+
normalized = max(0.0, min(1.0, (value - min_val) / (max_val - min_val)))
90
+
return scale_notes[int(normalized * (len(scale_notes) - 1))]
91
+
92
+
def value_to_velocity(self, value: float, min_val: float, max_val: float,
93
+
min_vel: int = 40, max_vel: int = 127) -> int:
94
+
if max_val == min_val:
95
+
return (min_vel + max_vel) // 2
96
+
normalized = max(0.0, min(1.0, (value - min_val) / (max_val - min_val)))
97
+
return int(min_vel + normalized * (max_vel - min_vel))
98
+
99
+
def value_to_cc(self, value: float, min_val: float, max_val: float) -> int:
100
+
if max_val == min_val:
101
+
return 64
102
+
normalized = max(0.0, min(1.0, (value - min_val) / (max_val - min_val)))
103
+
return int(normalized * 127)
104
+
105
+
def send_note_on(self, note: int, velocity: int = 100, channel: int = 0):
106
+
if self.port is None:
107
+
self.open()
108
+
self.port.send(mido.Message('note_on', note=note, velocity=velocity, channel=channel))
109
+
110
+
def send_note_off(self, note: int, channel: int = 0):
111
+
if self.port:
112
+
self.port.send(mido.Message('note_off', note=note, velocity=0, channel=channel))
113
+
114
+
def send_cc(self, cc: int, value: int, channel: int = 0):
115
+
if self.port is None:
116
+
self.open()
117
+
self.port.send(mido.Message('control_change', control=cc, value=value, channel=channel))
118
+
119
+
def send_program_change(self, program: int, channel: int = 0):
120
+
if self.port is None:
121
+
self.open()
122
+
self.port.send(mido.Message('program_change', program=program, channel=channel))
123
+
124
+
def play_note(self, note: int, duration: float, velocity: int = 100, channel: int = 0):
125
+
self.send_note_on(note, velocity, channel)
126
+
with self._lock:
127
+
self.active_notes[(channel, note)] = time.time() + duration
128
+
129
+
def note_off_later():
130
+
time.sleep(duration)
131
+
with self._lock:
132
+
self.active_notes.pop((channel, note), None)
133
+
self.send_note_off(note, channel)
134
+
135
+
threading.Thread(target=note_off_later, daemon=True).start()
136
+
137
+
def all_notes_off(self, channel: Optional[int] = None):
138
+
if not self.port:
139
+
return
140
+
for ch in ([channel] if channel is not None else range(16)):
141
+
self.send_cc(123, 0, ch)
142
+
143
+
@staticmethod
144
+
def list_ports() -> List[str]:
145
+
return mido.get_output_names()
146
+
147
+
148
+
class MIDIInstrument:
149
+
def __init__(self, output: MIDIOutput, channel: int = 0, program: int = 0,
150
+
scale: str = 'major', root: str = 'C', octave: int = 4):
151
+
self.output = output
152
+
self.channel = channel
153
+
self.program = program
154
+
self.scale = scale
155
+
self.root = root
156
+
self.octave = octave
157
+
self.scale_notes = output.get_scale_notes(root, scale, octave)
158
+
159
+
def setup(self):
160
+
self.output.send_program_change(self.program, self.channel)
161
+
162
+
def play_value(self, value: float, min_val: float, max_val: float, duration: float = 0.5):
163
+
note = self.output.value_to_note(value, min_val, max_val, self.scale_notes)
164
+
velocity = self.output.value_to_velocity(value, min_val, max_val)
165
+
self.output.play_note(note, duration, velocity, self.channel)
166
+
167
+
def send_cc_value(self, cc: int, value: float, min_val: float, max_val: float):
168
+
self.output.send_cc(cc, self.output.value_to_cc(value, min_val, max_val), self.channel)
+51
probe_display.py
+51
probe_display.py
···
1
+
# SPDX-License-Identifier: MIT
2
+
# Probe SysEx display addresses on a Roland GS synthesizer.
3
+
# Run with a MIDI port number: ./probe_display.py 1
4
+
5
+
import mido
6
+
import time
7
+
import sys
8
+
9
+
10
+
def gs_checksum(addr, data):
11
+
return (0x80 - (sum(addr) + sum(data)) % 0x80) % 0x80
12
+
13
+
14
+
def send_text(port, model_id, addr, text, device_id=0x10):
15
+
padded = text[:16].ljust(16)
16
+
data = [ord(c) & 0x7F for c in padded]
17
+
cs = gs_checksum(addr, data)
18
+
port.send(mido.Message('sysex', data=[0x41, device_id, model_id, 0x12] + list(addr) + data + [cs]))
19
+
20
+
21
+
ports = mido.get_output_names()
22
+
print("Available ports:")
23
+
for i, p in enumerate(ports):
24
+
print(f" {i}: {p}")
25
+
26
+
port_num = int(sys.argv[1]) if len(sys.argv) > 1 else 1
27
+
port = mido.open_output(ports[port_num])
28
+
print(f"\nUsing: {ports[port_num]}\n")
29
+
30
+
candidates = [
31
+
(0x42, (0x40, 0x00, 0x20), "GS 40 00 20"),
32
+
(0x42, (0x40, 0x00, 0x00), "GS 40 00 00"),
33
+
(0x42, (0x10, 0x00, 0x00), "GS 10 00 00"),
34
+
(0x42, (0x20, 0x00, 0x00), "GS 20 00 00"),
35
+
(0x42, (0x17, 0x10, 0x00), "GS 17 10 00"),
36
+
(0x45, (0x40, 0x00, 0x20), "45h 40 00 20"),
37
+
(0x45, (0x10, 0x00, 0x00), "45h 10 00 00"),
38
+
(0x4C, (0x40, 0x00, 0x20), "4Ch 40 00 20"),
39
+
(0x4C, (0x10, 0x00, 0x00), "4Ch 10 00 00"),
40
+
(0x42, (0x40, 0x00, 0x20), "GS 40 00 20 bcast"),
41
+
]
42
+
43
+
for i, (model, addr, desc) in enumerate(candidates):
44
+
dev_id = 0x7F if "bcast" in desc else 0x10
45
+
label = f"TEST {i+1}: {desc}"[:16]
46
+
print(f"Sending: {label!r} model=0x{model:02X} addr={[hex(b) for b in addr]}")
47
+
send_text(port, model, addr, label, device_id=dev_id)
48
+
time.sleep(2.0)
49
+
50
+
port.close()
51
+
print("Done — which test number appeared on the display?")
+183
snmp_client.py
+183
snmp_client.py
···
1
+
# SPDX-License-Identifier: MIT
2
+
3
+
from pysnmp.hlapi.asyncio import (
4
+
get_cmd, next_cmd, SnmpEngine, CommunityData, UdpTransportTarget,
5
+
ContextData, ObjectType, ObjectIdentity,
6
+
)
7
+
from typing import Dict, List, Any, Optional
8
+
9
+
10
+
class SNMPClient:
11
+
def __init__(self, host: str, port: int = 161, community: str = 'public',
12
+
version: int = 2, timeout: int = 5, retries: int = 3):
13
+
self.host = host
14
+
self.port = port
15
+
self.community = community
16
+
self.version = version
17
+
self.timeout = timeout
18
+
self.retries = retries
19
+
self.engine = SnmpEngine()
20
+
21
+
def _auth(self):
22
+
mp = {3: 2, 2: 1, 1: 0}.get(self.version, 1)
23
+
return CommunityData(self.community, mpModel=mp)
24
+
25
+
async def _transport(self):
26
+
return await UdpTransportTarget.create((self.host, self.port))
27
+
28
+
async def get(self, oids: List[str]) -> Dict[str, Any]:
29
+
objects = [ObjectType(ObjectIdentity(oid)) for oid in oids]
30
+
err_ind, err_status, _, var_binds = await get_cmd(
31
+
self.engine, self._auth(), await self._transport(), ContextData(), *objects
32
+
)
33
+
if err_ind:
34
+
raise Exception(f"SNMP error: {err_ind}")
35
+
if err_status:
36
+
raise Exception(f"SNMP error: {err_status.prettyPrint()}")
37
+
return {str(vb[0]): vb[1].prettyPrint() for vb in var_binds}
38
+
39
+
async def walk(self, oid: str) -> Dict[str, Any]:
40
+
results = {}
41
+
current = ObjectIdentity(oid)
42
+
transport = await self._transport()
43
+
while True:
44
+
err_ind, err_status, _, var_binds = await next_cmd(
45
+
self.engine, self._auth(), transport, ContextData(),
46
+
ObjectType(current), lexicographicMode=False,
47
+
)
48
+
if err_ind or err_status or not var_binds:
49
+
break
50
+
oid_str, value = var_binds[0]
51
+
if not str(oid_str).startswith(oid):
52
+
break
53
+
results[str(oid_str)] = value.prettyPrint()
54
+
current = ObjectIdentity(str(oid_str))
55
+
return results
56
+
57
+
async def _load_avg(self, index: int) -> Optional[float]:
58
+
oid = f'1.3.6.1.4.1.2021.10.1.3.{index}'
59
+
try:
60
+
result = await self.get([oid])
61
+
value = result.get(oid)
62
+
if value is None:
63
+
return None
64
+
return float(str(value).split(':')[-1].strip() if ':' in str(value) else value)
65
+
except Exception as e:
66
+
print(f" [DEBUG] load avg [{index}] error: {e}")
67
+
return None
68
+
69
+
async def get_cpu_load(self) -> Optional[float]:
70
+
return await self._load_avg(1)
71
+
72
+
async def get_cpu_load_5min(self) -> Optional[float]:
73
+
return await self._load_avg(2)
74
+
75
+
async def get_cpu_load_15min(self) -> Optional[float]:
76
+
return await self._load_avg(3)
77
+
78
+
async def get_memory_usage(self) -> Dict[str, int]:
79
+
oids = {
80
+
'total': '1.3.6.1.4.1.2021.4.5.0',
81
+
'available': '1.3.6.1.4.1.2021.4.6.0',
82
+
'cached': '1.3.6.1.4.1.2021.4.15.0',
83
+
'buffer': '1.3.6.1.4.1.2021.4.14.0',
84
+
}
85
+
try:
86
+
result = await self.get(list(oids.values()))
87
+
return {name: int(result.get(oid, 0)) for name, oid in oids.items()}
88
+
except Exception as e:
89
+
print(f" [DEBUG] Memory usage error: {e}")
90
+
return {}
91
+
92
+
async def get_disk_usage(self) -> List[Dict[str, Any]]:
93
+
base = '1.3.6.1.2.1.25.2.3.1'
94
+
try:
95
+
descrs = await self.walk(f'{base}.3')
96
+
sizes = await self.walk(f'{base}.5')
97
+
useds = await self.walk(f'{base}.6')
98
+
disks = []
99
+
for oid, name in descrs.items():
100
+
idx = oid.split('.')[-1]
101
+
size = int(sizes.get(f'{base}.5.{idx}', 0))
102
+
used = int(useds.get(f'{base}.6.{idx}', 0))
103
+
if size > 0:
104
+
disks.append({'name': name, 'size': size, 'used': used,
105
+
'percent': used / size * 100})
106
+
return disks
107
+
except:
108
+
return []
109
+
110
+
async def get_network_stats(self) -> List[Dict[str, Any]]:
111
+
base = '1.3.6.1.2.1.2.2.1'
112
+
try:
113
+
descrs = await self.walk(f'{base}.2')
114
+
in_octets = await self.walk(f'{base}.10')
115
+
out_octets = await self.walk(f'{base}.16')
116
+
return [
117
+
{
118
+
'name': name,
119
+
'in_bytes': int(in_octets.get(f'{base}.10.{oid.split(".")[-1]}', 0)),
120
+
'out_bytes': int(out_octets.get(f'{base}.16.{oid.split(".")[-1]}', 0)),
121
+
}
122
+
for oid, name in descrs.items()
123
+
]
124
+
except:
125
+
return []
126
+
127
+
async def get_process_count(self) -> int:
128
+
oid = '1.3.6.1.2.1.25.1.6.0'
129
+
try:
130
+
result = await self.get([oid])
131
+
value = result.get(oid)
132
+
return int(value) if value is not None else 0
133
+
except Exception as e:
134
+
print(f" [DEBUG] Process count error: {e}")
135
+
return 0
136
+
137
+
async def get_cpu_core_stats(self) -> Dict[str, int]:
138
+
oids = {
139
+
'cpu_user': '1.3.6.1.4.1.2021.11.50.0',
140
+
'cpu_nice': '1.3.6.1.4.1.2021.11.51.0',
141
+
'cpu_system': '1.3.6.1.4.1.2021.11.52.0',
142
+
'cpu_idle': '1.3.6.1.4.1.2021.11.53.0',
143
+
'cpu_iowait': '1.3.6.1.4.1.2021.11.54.0',
144
+
'cpu_kernel': '1.3.6.1.4.1.2021.11.55.0',
145
+
'cpu_interrupt': '1.3.6.1.4.1.2021.11.56.0',
146
+
'cpu_softirq': '1.3.6.1.4.1.2021.11.57.0',
147
+
'cpu_steal': '1.3.6.1.4.1.2021.11.58.0',
148
+
}
149
+
try:
150
+
result = await self.get(list(oids.values()))
151
+
return {
152
+
name: int(result[oid])
153
+
for name, oid in oids.items()
154
+
if oid in result and 'No Such Object' not in str(result[oid])
155
+
}
156
+
except Exception as e:
157
+
print(f" [DEBUG] CPU stats error: {e}")
158
+
return {}
159
+
160
+
async def get_per_core_cpu_load(self) -> Dict[str, int]:
161
+
oid = '1.3.6.1.2.1.25.3.3.1.2'
162
+
results = {}
163
+
try:
164
+
for oid_str, value in sorted((await self.walk(oid)).items()):
165
+
core_num = int(oid_str.split('.')[-1]) - 196608
166
+
if 0 <= core_num < 32:
167
+
results[f'core{core_num}'] = int(value)
168
+
except Exception as e:
169
+
print(f" [DEBUG] Per-core CPU load error: {e}")
170
+
return results
171
+
172
+
async def get_cpu_load_alternative(self) -> Optional[float]:
173
+
try:
174
+
result = await self.walk('1.3.6.1.2.1.25.3.3.1.2')
175
+
if result:
176
+
values = [float(v) for v in result.values()]
177
+
return sum(values) / len(values) if values else None
178
+
except:
179
+
pass
180
+
return None
181
+
182
+
def close(self):
183
+
pass