drivers.CubicPM3006.CubicPM3006.CubicPM3006

class drivers.CubicPM3006.CubicPM3006.CubicPM3006(cubic_port: str = '/dev/ttyAMA1', baud: int = 9600)

UART driver for the Cubic PM3006 laser particulate matter sensor.

Wraps a pyserial connection to a Cubic PM3006 device and exposes helpers to request and decode its binary protocol. The sensor reports mass concentrations for PM1, PM2.5, PM4.25, PM10, and TSP (referred to here as pm100) in micrograms per cubic metre, together with particle counts in six size bins (>0.3, >0.5, >1.0, >2.5, >5.0, and >10 micrometres).

While this class is self-contained, it is functionally a sibling of drivers.CubicGeneric.CubicGeneric.CubicGeneric and follows the same conventions as other sensor drivers built on top of SensorBase.SensorBase.GenericSensor.

Variables:

  • CUBIC_READ_DATA (list[int]) – Command byte sequence used to request a PM/particle-count measurement frame from the sensor.

  • DEBUG (bool) – When True, enables verbose serial-level logging of every TX and RX transaction.

  • data (dict[str, float]) – Most recent decoded readings. Keys are pm1, pm2_5, pm4_25, pm10, pm100 (mass concentrations) and p1 through p6 (particle counts).

  • TIMEOUT (int) – Read timeout in milliseconds for a complete frame.

  • COMMAND_DELAY (int) – Seconds to wait between consecutive commands.

  • RESPONSE_DELAY (float) – Seconds to wait after writing before polling the input buffer.

  • MAX_BUFFER_SIZE (int) – Expected length, in bytes, of a full response frame.

Example

Instantiate and read one sample on a Raspberry Pi:

>>> from drivers.CubicPM3006.CubicPM3006 import CubicPM3006
>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> sensor.getPM(sendCommand=True)["pm10"]
18.0

Initialise the UART connection and runtime state for the sensor.

Opens the underlying pyserial port with the PM3006’s required 8N1 framing, primes the read-data command, and zero-initialises the data cache so callers can rely on a fully-populated dictionary even before the first measurement is taken.

Parameters:

  • cubic_port (str) – Path to the serial device the PM3006 is attached to. Defaults to "/dev/ttyAMA1" which is the standard UART on Oizom Raspberry Pi based monitors.

  • baud (int) – Baud rate for the UART link. The PM3006 firmware fixes this at 9600 bps; only override when bridging through hardware that demands a different rate.

Returns:

The constructor mutates self and does not return a value.

Return type:

None

Raises:

  • serial.SerialException – If the underlying serial port cannot be opened (for example, the device file does not exist, permissions are missing, or the port is in use).

  • ValueError – If baud is not a value accepted by pyserial.

Example

Open the sensor on the default UART:

>>> from drivers.CubicPM3006.CubicPM3006 import CubicPM3006
>>> sensor = CubicPM3006("/dev/ttyAMA1")

Note

The serial port is held open for the lifetime of this instance. Callers that need to share the UART with other drivers should serialise access externally. Unlike sensors built on SensorBase.SensorBase.GenericSensor, this driver does not register itself with a global sensor manager.

getPM(sendCommand: bool = False) → dict[str, float]

Read particulate-matter mass concentrations and particle counts.

When sendCommand is True the driver transmits CUBIC_READ_DATA, waits for the sensor response, and decodes the binary frame into mass concentrations and particle counts. PM fields are reconstructed from 32-bit big-endian integers at fixed offsets in the response payload. When sendCommand is False the cached data dictionary is returned unchanged.

Parameters:

sendCommand (bool) – If True, issue a read command before returning. If False, return the cached values from the previous successful read.

Returns:

Dictionary with the following keys.

  • pm1 – PM1.0 mass concentration (ug/m^3).

  • pm2_5 – PM2.5 mass concentration (ug/m^3).

  • pm4_25 – PM4.25 mass concentration (ug/m^3).

  • pm10 – PM10 mass concentration (ug/m^3).

  • pm100 – TSP/PM100 mass concentration (ug/m^3).

  • p1 – particle count for >0.3 um bin.

  • p2 – particle count for >0.5 um bin.

  • p3 – particle count for >1.0 um bin.

  • p4 – particle count for >2.5 um bin.

  • p5 – particle count for >5.0 um bin.

  • p6 – particle count for >10 um bin.

Return type:

dict[str, float]

Raises:

  • Exception – Propagated from sendCommand() when the sensor fails to respond within self.TIMEOUT milliseconds. In practice sendCommand() catches serial errors internally, so a malformed or missing response simply results in zero-valued entries.

  • ValueError – If the response contains non-hexadecimal byte strings that int(data, 16) cannot parse.

Example

Trigger a fresh measurement and inspect PM2.5:

>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> reading = sensor.getPM(sendCommand=True)
>>> reading["pm2_5"]
12.0

Note

Mass-concentration values are decoded from a 32-bit big-endian integer b3 * 256^3 + b2 * 256^2 + b1 * 256 + b0 where b3..b0 are sequential payload bytes. Particle-count decoding uses the same layout but starts at offset 27. A valid frame is at least 16 bytes long and starts with the 0x16 header byte; shorter or differently-headed frames leave the corresponding entries at their zero defaults. Mirrors the conventions used by drivers.CubicGeneric.CubicGeneric.CubicGeneric.

get_serial_number() → str

Read the factory-programmed serial number from the sensor.

Issues the SERIAL_NUMBER command, waits for the response frame, and reconstructs the device serial number from five consecutive 16-bit big-endian words. Each word is rendered as a zero-padded 4-digit decimal and the segments are concatenated to form a single 20-character string.

Parameters:

  • None – This method takes no positional or keyword arguments

  • self. (beyond)

Returns:

The decoded serial number as a numeric string. Returns an empty string when the response is shorter than expected or does not start with the 0x16 header byte.

Return type:

str

Raises:

  • AttributeError – If self.SERIAL_NUMBER is not defined on the instance. This driver does not declare a default serial-number command; callers are expected to inject one or subclass before invoking this method.

  • Exception – Propagates any underlying serial timeout raised by sendCommand() if the sensor fails to reply within self.TIMEOUT milliseconds.

Example

Read the serial number from a connected sensor:

>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> sensor.SERIAL_NUMBER = [0x11, 0x01, 0x1F, 0xCF]
>>> sensor.get_serial_number()
'00010002000300040005'

Note

The PM3006 returns its serial number as five 16-bit big-endian fields starting at byte index 3 of the response payload. The 4-digit zero padding preserves leading zeros so that the assembled string is always the same length for a given device family.

micros() → int

Return the current wall-clock time in microseconds.

Convenience helper that scales time.time() to microseconds so that timing logic in this driver reads similarly to the Arduino reference implementations the protocol was first ported from.

Parameters:

None – This method takes no arguments beyond self.

Returns:

Integer microsecond timestamp since the Unix epoch.

Return type:

int

Raises:

None – This method does not raise under normal operation; it inherits any failure modes of time.time().

Example

Capture a microsecond timestamp:

>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> isinstance(sensor.micros(), int)
True

Note

Resolution is limited to the host clock granularity (typically on the order of a microsecond on Linux). This helper is not monotonic; for elapsed-time measurements that must not be affected by NTP adjustments, prefer time.monotonic_ns().

millis() → int

Return the current wall-clock time in milliseconds.

Used internally by sendCommand() to implement the TIMEOUT watchdog while polling for sensor responses.

Parameters:

None – This method takes no arguments beyond self.

Returns:

Integer millisecond timestamp since the Unix epoch.

Return type:

int

Raises:

None – This method does not raise under normal operation; it inherits any failure modes of time.time().

Example

Capture a millisecond timestamp:

>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> isinstance(sensor.millis(), int)
True

Note

Mirrors the Arduino millis() API. Not monotonic; prefer time.monotonic() if the timestamps must survive system clock changes.

sendCommand(command, recieve: bool = True) → list[str]

Transmit a raw command frame and collect the response bytes.

Re-opens the serial port if it has been closed, writes each command byte individually, waits RESPONSE_DELAY seconds, and then drains the input buffer one byte at a time. Every received byte is recorded both as a raw bytes object (for internal debug logging) and as a Python hex string (returned to the caller). The polling loop respects self.TIMEOUT and raises if the sensor does not start responding in time.

Parameters:

  • command (Iterable[int]) – Iterable of byte values (0-255) to transmit to the sensor. Each value is written as a single-byte bytes object.

  • recieve (bool) – Legacy flag retained for API compatibility. Currently only controls debug-side behaviour and does not close the port between transactions; preserved to keep downstream callers working unchanged.

Returns:

Hex-string representations of every byte received from the sensor (for example ["0x16", "0x2d", ...]). Returns whatever was accumulated so far if an exception is caught internally, which may be an empty list.

Return type:

list[str]

Raises:

  • Exception – A generic Exception with the message "[ERR CUBIC] Timeout CUBIC" is raised internally and caught when the sensor produces no data within self.TIMEOUT milliseconds; the caught exception is logged via the context logger.

  • serial.SerialException – Propagated if the underlying serial port fails to (re)open or a write operation errors. The except block closes the port and returns whatever bytes were captured.

Example

Send the standard read-data command and inspect the response:

>>> sensor = CubicPM3006("/dev/ttyAMA1")
>>> sensor.sendCommand(sensor.CUBIC_READ_DATA)
['0x16', '0x2d', '0xb', ...]

Note

The PM3006 expects the entire command frame within a tight window, so this method enforces a 100 ms pre-write delay and a 10 ms inter-byte delay on receive. Enable self.DEBUG for a TX/RX trace through utils.oizom_logger.OizomLogger.

COMMAND_DELAY = 1
CUBIC_READ_DATA = [b'\x11', b'\x02', b'\x0b', b'\x07', b'\xdb']
DEBUG = False
MAX_BUFFER_SIZE = 51
RESPONSE_DELAY = 0.1
SERIAL_NUMBER = [17, 1, 31, 207]
TIMEOUT = 8000
__cubic_port
_timeout_ = 8
data