drivers.Noise.Noise.Noise

class drivers.Noise.Noise.Noise

SAMD-based sound pressure level (SPL) / dBA meter driver.

This class drives the legacy Oizom noise board: a calibrated MEMS microphone whose analog signal is digitised, A-weighted and integrated by a SAMD microcontroller. The MCU answers each NOISE?\r\n request with a fixed 8-byte binary frame in the layout [0x01, 0x03, 0x00, LAeq_H, LAeq_L, LZeq_H, LZeq_L, 0x0A], where the two 16-bit big-endian fields encode the latest LAeq and LZeq values in tenths of a decibel.

What it measures:

  • LAeq – A-weighted equivalent continuous sound pressure level in dB(A) SPL. This is the canonical “noise” reading consumed by the dashboard / API and is the human-perceived loudness metric used by most environmental regulations.

  • LZeq – Un-weighted (flat / Z-weighting) equivalent continuous sound level in dB SPL. Useful for low-frequency content that A-weighting attenuates.

The driver also exposes the standard Oizom sensor lifecycle (initialize(), getSensorReading(), putSensorValue()), mirroring the contract defined by SensorBase.SensorBase.GenericSensor.

Variables:

  • FRAME_LEN (int) – Expected response frame length in bytes (always 8).

  • TIMEOUT (int) – Maximum wait time for a complete frame, in milliseconds, before declaring a read failure.

  • ser (serial.Serial | None) – Open pyserial instance used for UART communication. None until initialize() is called.

  • configuration (dict) – Sensor configuration dict supplied by the scheduler (enable flag, part number, debug flag, parameters list mapping pm codes to short-codes).

  • LAeq (list[float]) – Accumulated A-weighted Leq samples (dB(A)) collected since the last putSensorValue() flush; used for energy averaging.

  • LZeq (list[float]) – Accumulated Z-weighted Leq samples (dB) collected since the last putSensorValue() flush.

  • debug (bool) – When True, verbose UART / frame tracing is emitted through the OizomLogger context logger.

  • part_number (int) – Hardware part-number identifier (defaults to 41) used by the platform to disambiguate sensor variants.

  • EOL (bytes) – Line terminator appended to ASCII commands (b"\r\n").

Example

>>> import serial
>>> from drivers.Noise.Noise import Noise
>>> ser = serial.Serial("/dev/ttyACM0", 115200, timeout=2)
>>> sensor = Noise()
>>> sensor.initialize(ser, {"en": 1, "debug": False})
True
>>> sensor.get_noise()
{'LAeq': 54.3, 'LZeq': 58.1}

Create a Noise driver with safe, un-configured defaults.

Sets the binary protocol constants (8-byte frame, 1600 ms timeout), clears the accumulation buffers for LAeq / LZeq, and assigns the default Oizom noise part-number (41). The instance is not yet usable for I/O at this point: the serial port and runtime configuration are attached later by initialize().

Parameters:

None.

Returns:

Constructors do not return a value.

Return type:

None

Raises:

None – Pure attribute assignment; no I/O is performed.

Example

>>> from drivers.Noise.Noise import Noise
>>> sensor = Noise()
>>> sensor.FRAME_LEN
8

Note

Mirrors the parameter-less constructor convention used across Oizom drivers and by SensorBase.SensorBase.GenericSensor, so the scheduler can instantiate any driver without knowing its specifics. All runtime state (serial port, debug, part_number, parameters) is bound later in initialize().

energy_average() → float

Reduce the accumulated LAeq samples to an acoustically correct mean.

Computes the equivalent continuous sound level over the aggregation window using the standard energy-averaging formula 10 * log10(mean(10**(dB / 10))). To prevent overflow when summing many high-dB samples the implementation subtracts the maximum value from each sample before exponentiating and adds it back at the end (a numerically stable log-sum-exp trick). If any of the intermediate arithmetic raises – typically because the buffer contained no valid samples – the method falls back to a plain arithmetic mean of LAeq.

Parameters:

None – The method reads from LAeq directly.

Returns:

The energy-averaged LAeq in dB(A), rounded to two decimal places. Returns 0.0 when the computed linear energy average is non-positive.

Return type:

float

Raises:

Example

>>> sensor = Noise()
>>> sensor.LAeq = [54.3, 55.1, 53.9, 60.0]
>>> sensor.energy_average()
56.31

Note

Arithmetic averaging of decibels is acoustically wrong: doubling the sound power corresponds to a 3 dB increase, not a doubling of the dB number. Always prefer this method over mean(LAeq) when reporting an Leq for an interval. The fallback exists only so the scheduler keeps emitting something if the buffer ever ends up in a pathological state.

getSensorReading() → dict

Return an instantaneous reading keyed by configured short-codes.

Invokes get_noise() to refresh LAeq / LZeq, then iterates the "parameters" list in configuration and emits one entry per parameter using its short-code ("sc") as the key. The semantics of each parameter are driven by its "pm" (parameter mode) code:

  • pm == 1 – instantaneous LAeq from the latest poll.

  • pm == 2 – running maximum of all LAeq samples in the current accumulation window.

  • pm == 3 – running minimum of all LAeq samples in the current accumulation window.

Parameters:

None.

Returns:

Mapping of parameter short-code (str) -> reading value (float). Empty when polling fails, when no configured parameters match, or when an exception is caught.

Return type:

dict

Raises:

  • None – Errors are logged via

  • OizomLogger.error_with_context –

  • empty-dict return so the scheduler can continue. –

Example

>>> sensor = Noise()
>>> sensor.initialize(
...     ser,
...     {"en": 1, "parameters": [{"sc": "no", "pm": 1}]},
... )
True
>>> sensor.getSensorReading()
{'no': 54.3}

Note

This method is the per-tick read interface expected by SensorBase.SensorBase.GenericSensor. It is intentionally forgiving – a single dropped frame results in an empty dict rather than an exception, so a misbehaving noise board never blocks the rest of the sensor stack. Long-running aggregates (energy-averaged Leq, min, max over the full window) are emitted by putSensorValue().

get_noise() → dict | None

Poll the sensor once and append the result to the rolling buffers.

Convenience wrapper that ties send_command() and parse_noise_response() together: it confirms a serial port has been bound, transmits the default NOISE? request, validates the returned frame, and – on success – appends the LAeq and LZeq samples to LAeq / LZeq so that putSensorValue() can later compute an energy-averaged Leq over the aggregation window.

Parameters:

None.

Returns:

{"LAeq": <float dB(A)>, "LZeq": <float dB>} for the current instantaneous reading, or None if the port is not initialised, the UART read failed, or the frame did not validate.

Return type:

dict | None

Raises:

  • None – Any unexpected exception is logged through

  • OizomLogger.error_with_context –

  • None` return so the scheduler can safely retry on the nex –

  • tick. –

Example

>>> sensor = Noise()
>>> sensor.initialize(ser, {"en": 1})
True
>>> sensor.get_noise()
{'LAeq': 54.3, 'LZeq': 58.1}

Note

This is the canonical per-tick entry point for the noise driver. Each successful call appends one sample to the internal buffers; those buffers are consumed and cleared by putSensorValue(). Because dB is a logarithmic unit, the LAeq accumulator must be reduced with energy_average() rather than an arithmetic mean.

initialize(serial_port, configuration: dict) → bool

Bind a UART port and configuration dict to the noise driver.

Stores the supplied serial.Serial instance and configuration on the instance, latches the debug flag and hardware part_number, and decides whether the sensor should be considered enabled based on the "en" key. Any exception during binding is caught and reported via OizomLogger so that a misbehaving sensor never tears down the scheduler.

Parameters:

  • serial_port (serial.Serial) – An already-opened pyserial port (typically /dev/ttyACM0 at 115200 baud) over which the NOISE? ASCII commands will be sent.

  • configuration (dict) –

    Per-sensor configuration. Recognised keys:

    • "en" (int): Enable flag; sensor is activated when truthy.

    • "debug" (bool, optional): Enable verbose UART tracing. Defaults to False.

    • "pn" (int, optional): Hardware part-number override. Defaults to 41.

    • "parameters" (list[dict], optional): Parameter mapping consumed by getSensorReading() and putSensorValue().

Returns:

True when the sensor is enabled ("en" truthy) and the configuration was bound without error, False otherwise (including when an exception is caught).

Return type:

bool

Raises:

None – All exceptions are caught internally and surfaced via the

:raises error-context logger; callers only need` to :py:class:`inspect the boolean: :raises return value.:

Example

>>> import serial
>>> from drivers.Noise.Noise import Noise
>>> ser = serial.Serial("/dev/ttyACM0", 115200, timeout=2)
>>> sensor = Noise()
>>> sensor.initialize(ser, {"en": 1, "pn": 41, "debug": True})
True

Note

Follows the lifecycle entry-point contract documented on SensorBase.SensorBase.GenericSensor: every Oizom driver exposes an initialize(serial_port, configuration) method that returns a boolean so the scheduler can prune dead sensors at boot. The serial port is not opened here – it must already be open before being passed in.

parse_noise_response(frame: bytes) → dict | None

Decode an 8-byte SAMD frame into float LAeq and LZeq dB values.

Validates that the frame is non-empty, has exactly FRAME_LEN bytes, and matches the expected [0x01, 0x03, 0x00, LAeq_H, LAeq_L, LZeq_H, LZeq_L, 0x0A] envelope (start 0x01 0x03, terminator 0x0A). The two 16-bit big-endian payload words are reconstructed and divided by 10 because the MCU transmits decibels in tenths to avoid floating point on the wire.

Parameters:

frame (bytes) – The raw 8-byte response frame as returned by send_command().

Returns:

A mapping {"LAeq": <float dB(A)>, "LZeq": <float dB>} on a well-formed frame, or None if the frame is empty, the wrong length, or fails the header / terminator check.

Return type:

dict | None

Raises:

  • None – All validation failures are reported as None returns

  • (with optional debug-level log lines); the method itself does –

  • not raise. –

Example

>>> sensor = Noise()
>>> raw = bytes([0x01, 0x03, 0x00, 0x02, 0x1F, 0x02, 0x45, 0x0A])
>>> sensor.parse_noise_response(raw)
{'LAeq': 54.3, 'LZeq': 58.1}

Note

Decoded values are short-window equivalent continuous sound levels (Leq). LAeq is A-weighted – the perceptual filter specified in IEC 61672 that mimics human hearing – while LZeq is the unweighted (flat) reference level. Both come already integrated by the SAMD firmware, so no further DSP is required on the Pi side.

putSensorValue(result: dict | None = None) → dict

Aggregate the current window into final values and reset the buffers.

Iterates the configured parameters and, for each one, derives the aggregate appropriate to its pm (parameter mode) code from the accumulated LAeq samples:

  • pm == 1 – energy-averaged LAeq (see energy_average()).

  • pm == 2 – maximum LAeq observed in the window.

  • pm == 3 – minimum LAeq observed in the window.

If the window is empty (no successful polls between calls) the last cached values (self.old_laeq, self.old_max_laeq, self.old_min_laeq) are re-emitted so the upstream pipeline always receives a numeric value. After populating result the LAeq and LZeq buffers are cleared in preparation for the next aggregation period.

Parameters:

result (dict | None) – Optional pre-existing accumulator to merge into – useful when several drivers share a single payload dict. When None (the default) a new empty dict is created.

Returns:

The (possibly newly created) result dict, populated with one entry per configured parameter keyed by its short-code.

Return type:

dict

Raises:

None – The method does not raise under normal conditions; callers

:raises should treat any exception coming out` of :py:class:`it as a bug.:

Example

>>> sensor = Noise()
>>> sensor.initialize(
...     ser,
...     {
...         "en": 1,
...         "parameters": [
...             {"sc": "no", "pm": 1},
...             {"sc": "no_max", "pm": 2},
...             {"sc": "no_min", "pm": 3},
...         ],
...     },
... )
True
>>> for _ in range(5):
...     sensor.getSensorReading()
>>> sensor.putSensorValue()
{'no': 56.31, 'no_max': 60.0, 'no_min': 53.9}

Note

This is the end-of-window companion to getSensorReading() and corresponds to the putSensorValue hook expected by SensorBase.SensorBase.GenericSensor. Because the buffers are cleared on exit, callers should treat the returned dict as the authoritative summary for the just-finished aggregation window – subsequent calls will reflect a fresh window.

send_command(command: bytes = b'NOISE?\r\n', delay: float = 0.5) → bytes | None

Write a UART command to the SAMD MCU and read back one 8-byte frame.

Flushes the input/output buffers (re-opening the port if it has been closed), writes command, waits delay seconds for the microcontroller to compute and emit a frame, then performs a chunked read until either FRAME_LEN bytes have arrived or TIMEOUT milliseconds have elapsed. The port is always closed in the finally block to keep the file descriptor count bounded between sampling ticks.

Parameters:

  • command (bytes) – Raw bytes to transmit on the UART link. Defaults to b"NOISE?\r\n", the only command the SAMD firmware currently recognises.

  • delay (float) – Seconds to sleep after the write before draining the receive buffer. 0.5 s comfortably covers the MCU’s Leq integration + transmission latency at 115200 baud.

Returns:

The raw 8-byte response frame on success, or None if the receive buffer was empty after delay, the read did not complete within TIMEOUT, or any underlying pyserial call raised.

Return type:

bytes | None

Raises:

  • None – All exceptions are caught and logged via

  • OizomLogger.error_with_context –

  • None` on any failure so the scheduler can simply skip th –

  • tick. –

Example

>>> sensor = Noise()
>>> sensor.initialize(ser, {"en": 1})
True
>>> frame = sensor.send_command()
>>> len(frame) if frame else 0
8

Note

delay directly caps the achievable sampling rate. With the default of 0.5 s the driver polls at roughly 2 Hz, which is more than sufficient for environmental Leq monitoring but is far below the audio Nyquist rate – all spectral analysis happens on the MCU. The serial port is closed at the end of every call, so callers may share the same serial.Serial instance with other drivers without leaking the lock.

EOL = b'\r\n'
FRAME_LEN = 8
LAeq: list[float] = []
LZeq: list[float] = []
TIMEOUT = 1600
configuration: dict
debug = False
part_number = 41
ser = None