PMS5003 Particulate Matter Sensor Test Run

Hi

In this article we’ll test out the PMS5003 sensor in order to see if it works. I’ve forgot to buy a connector board, so we will do a manual connection to the Raspberry Pi 3 B V2. This involves cutting the wires and adding some resistors.

Please note that you need:

  • 5 Jumper Wires
  • 2 Resistors 10K Ohm

Raspberry Pi Setup

Before connecting the sensor to the Pi we need to configure the Pi for this usecase.

Note that if you’re using this sensor with Raspberry Pi, then you’ll need to make a couple of changes to its configuration. Type sudo raspi-config in the terminal and then under “Interfacing options” and “Serial” disable the login shell and enable the serial port hardware. Edit your /boot/config.txt file and add the lines enable_uart=1 and dtoverlay=pi3-miniuart-bt to the bottom of the file.

From: https://shop.pimoroni.com/products/pms5003-particulate-matter-sensor-with-cable

After the setup from above is done, reboot the Pi and install the software.

sudo pip install pms5003
git clone https://github.com/pimoroni/pms5003-python

Note: In order to connect to the Raspberry Pi I’ve used Visual Studio Code and the Remote – SSH, this is unnecessary, editing files is VSCode is a personal preference.

Sensor Wiring

To identify the sensor’s wires I’ve consulted the PMS5003 specification and I’ve cut my sensor’s Picoblade connector, this enabled me to use jumper wires and attach my 10K ohms resistors to the wires. Then I’ve isolated the wires using some electrical tape.

Note: I’ve connected PIN 3 and PIN 6 to a 10K ohm pull up resistor at 3.3V because at the moment I don’t know if I need to reset the sensor or put it to sleep.

To simplify things further:

SENSOR 1 (VCC) -> RPI 5V

SENSOR 2 (GND) -> RPI Ground

SENSOR 3 & 6 (SET & RESET) -> 10k OHM -> RPI 3.3V

SENSOR 4 (RXD) -> RPI Gpio 14 (UART TX)

SENSOR 6 (TXD) -> RPI Gpio 15 (UART RX)

That’s it! It will look something like the following.

Verify that the sensor is connected then ssh into the Pi and run the all.py example from pms5003-python. You should see something like:

The results can be interpreted using the following reference table.

Thanks for reading!

Context Managers and Cross Cutting concerns in Python

Hello,

In this short article I would like to talk about context managers. I personally consider that at the core they are just a form of decorators. If you don’t know what a decorator is check the Decorator Pattern Wikipedia article.

Decorators can be used to implement cross-cutting concerns. We have componentA and we need logging and security, we could write the logic for logging and security handling in componentA but some people consider component a should be componentA not componentAthatAlsoKnowsAboutSecurityAndOtherStuff. Since it’s not the component’s responsibility to authorize requests or log calls to a external logging service, we can wrap the componentA into a decorator that does just that.

A formal definition for cross-cutting concerns as taken from Wikipedia is the following:

In aspect-oriented software development, cross-cutting concerns are aspects of a program that affect other concerns. These concerns often cannot be cleanly decomposed from the rest of the system in both the design and implementation, and can result in either scattering (code duplication), tangling (significant dependencies between systems), or both.

And some examples of cross cutting concerns include:

Since the context managers are sort of similar to decorators you can use them to implement cross cutting concerns. Let’s explore.

Simple Example

In Python you can have two types of context managers: a function and a class. In order for the function to behave like a context manager it will need to be decorated with the @contextmanager decorator, and in order for a class behave like a context manager it needs to implement __enter__ and __exit__.

Context managers can be called using the with statement. The following code snippet demonstrates two context managers:

  • One that logs when the function is called and when it exits.
  • One that intercepts the function arguments.
from contextlib import contextmanager

@contextmanager
def simple_context_manager(function):
    try:
        print("calling function")
        yield function
    finally:
        print("function call has ended")

class SimpleContextManager:
    def __init__(self, cb):
        self.cb = cb

    def _intercept(self, *args, **kwargs):
        print(f"calling with {args} {kwargs}")
        return print(*args, **kwargs)

    def __enter__(self):
        print("intercept start")
        return self._intercept

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("intercept end")

def main():
    with simple_context_manager(print) as print_func:
        print_func("hi")

    with SimpleContextManager(print) as print_func:
        print_func("hi")
        print_func("hi", end="\n\n", sep=",")
        print_func("hi")

if __name__ == '__main__':
    main()

Caching

What is caching? In short..

Caching is used to store the result of an expensive computation somewhere in memory or on a persistent storage device in order to optimize the program.

We have the compute_fibonacci function, which is quite slow. A version that uses cache has been implementing in the CachedComputeFibonacci class. Notice how the code takes some time to output the result for the first call of print(cached_compute_fibonacci(35)) statement but the second print in instant.

def compute_fibonacci(number):
    if number <= 1:
        return number
    return compute_fibonacci(number-1) + compute_fibonacci(number-2)


class CachedComputeFibonacci:
    def __init__(self):
        self._cache = {}

    def __call__(self, *args, **kwargs):
        number = args[0]
        if number in self._cache:
            return self._cache[number]
        result = compute_fibonacci(number)
        self._cache[number] = result
        return result

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

def main():
    # Non cached
    print(compute_fibonacci(10))

    # Cached
    with CachedComputeFibonacci() as cached_compute_fibonacci:
        print(cached_compute_fibonacci(35))
        print(cached_compute_fibonacci(35))



if __name__ == '__main__':
    main()

Logging

Logging can be useful for debugging and auditing purposes.

def compute_fibonacci(number):
    if number <= 1:
        return number
    return compute_fibonacci(number-1) + compute_fibonacci(number-2)


class LoggedComputeFibonacci:
    def __init__(self):
        pass

    def __call__(self, *args, **kwargs):
        print(f"calling compute_fibonacci with args={args} kwargs={kwargs}")
        result = compute_fibonacci(args[0])
        print(f"compute_fibonacci={result}")
        return result

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

def main():
    # Logging
    with LoggedComputeFibonacci() as cached_compute_fibonacci:
        print(cached_compute_fibonacci(35))
        print(cached_compute_fibonacci(36))



if __name__ == '__main__':
    main()

Error detection and correction

If you find yourself duplicating the same try/catch logic in multiple places of your code perhaps you can extract it into a context manager for handling errors:

from contextlib import contextmanager

@contextmanager
def my_error_handler():
    try:
        yield
    except ZeroDivisionError:
        print("abort abort")

def main():
    # error handling
    with my_error_handler():
        print("0 / 0 =", 0 / 0)



if __name__ == '__main__':
    main()

The code is definitely more cleaner this way, in my opinion.

Thanks for reading and I hope that you’ve learnt something!