Use Cases

Assisted Living

An area of interest in Internet of Things is the need for Smart Home solutions that are designed for people with disability and mobility needs in order to provide independent Assisted Living.

This could be accomplished through sensors that trigger events with minimal or unconventional means of interaction by the end user, such as through voice, controllers, or simply human presence. Actuators, such as plugs, locks, kitchen appliances, and alarms, can trigger as a consequence of these minimal human interactions. Assisted Living can leverage the low-latency event detection and actuation from BoboCEP to ensure a reliable and rapid response to events.

Note

The example below is for illustrative purposes only.

Phenomenon

Consider a single phenomenon that one might wish to detect and automatically respond to in an Assisted Living solution. For example, if an elderly resident were to fall in their home or living facility and require immediate assistance, then this ‘phenomenon’ could be detected through one or more different patterns of correlated, temporal data.

Code

phenom_fall = BoboPhenomenon(
    name="Fall",
    patterns=[pattern_1, pattern_2],
    action=action_fall
)

Pattern #1

A person is detected entering the living room and they have not been detected moving for at least 60 seconds, nor detected in any other room.

Code

from bobocep.cep.phenom import BoboPattern, BoboPatternBuilder

# Living room motion starts the pattern, where event data is a dict that
# contains keys "room" and "motion".
fb1 = lambda e, h: e.data["room"] == "living_room" and e.data["motion"] == 1

# No motion in living room after 60 seconds. The duration between the event
# accepted during fb1 and the event being evaluated with fb2 is used to
# determine duration, where timestamp is milliseconds since the epoch.
fb2 = lambda e, h: e.data["room"] == "living_room" and e.data["motion"] == 0 \
                   and (e.timestamp - h.first().timestamp) / 1000 >= 60

# Any motion in any room halts the pattern.
hc1 = lambda e, h: e.data["motion"] == 1

my_pattern: BoboPattern = BoboPatternBuilder("pattern_1") \
    .followed_by(fb1) \
    .followed_by(fb2) \
    .haltcondition(hc1) \
    .generate()

Pattern #2

A heart-rate sensor was operating at healthy levels (for example, between 60 to 140 beats per minute), but has started to consistently report values outside of this range.

Code

from bobocep.cep.phenom import BoboPattern, BoboPatternBuilder

# Initial indication that heart-rate sensor operates within normal levels
fb1 = lambda e, h: 60 <= int(e.data) <= 140

# Abnormal heart rates. Using the `times` parameter in the `followed_by`
# method below, 10 events that match fb2 must occur in sequence.
fb2 = lambda e, h: not(60 <= int(e.data) <= 140)

# Halts if still operating within normal levels.
hc1 = lambda e, h: 60 <= int(e.data) <= 140

my_pattern: BoboPattern = BoboPatternBuilder("pattern_1") \
    .followed_by(fb1) \
    .followed_by(fb2, times=10) \
    .haltcondition(hc1) \
    .generate()

Actions

On fulfilment of the phenomenon via any of its patterns, a complex event is generated and one or more actions may be triggered.

For this, we could use the BoboActionMultiSequential action, which takes multiple actions and runs them sequentially. It can attempt to run them all in sequence and continue execution even if some of them were to fail. This is useful for our scenario because we can trigger several actions for redundancy. For example, we can notify multiple neighbours of the emergency even if some requests failed to send.

Code

action_fall = BoboActionMultiSequential(
    name="Action Fall",
    actions=[action_unlock_door, action_notify_neighbours],
    stop_on_fail=False
)

Action #1

Unlock the front door, to allow for easy access by neighbours, care workers, or paramedics.

Below is a custom action, BoboActionIFTTTWebhooks, that uses the IFTTT Webhooks Integration to accomplish this. A Webhooks request can trigger various smart locks that are integrated into the IFTTT service. For example, Kubu or Nuki.

The webhooks_event_name parameter is the custom Event Name that is entered when setting up the Webhooks integration. The webhooks_key is provided in the Webhooks Documentation that appears here after making an event.

Code

from typing import Tuple, Any
from bobocep.cep.action import BoboAction
from bobocep.cep.event import BoboEventComplex
import requests


class BoboActionIFTTTWebhooks(BoboAction):
    """
    An action that triggers an event using the IFTTT Webhooks integration.
    See: https://ifttt.com/maker_webhooks
    """

    _URL = "https://maker.ifttt.com/trigger/{0}/json/with/key/{1}"

    def __init__(
            self,
            name: str,
            webhooks_event_name: str,
            webhooks_key: str,
            *args,
            **kwargs):
        """
        :param name: The action name.
        :param webhooks_event_name: IFTTT Webhooks event name.
        :param webhooks_key: IFTTT Webhooks key.
        :param args: Action arguments.
        :param kwargs: Action keyword arguments.
        """
        super().__init__(name=name, args=args, kwargs=kwargs)

        self._webhooks_event_name = webhooks_event_name
        self._webhooks_key = webhooks_key

    def execute(self, event: BoboEventComplex) -> Tuple[bool, Any]:
        """
        :param event: The complex event that triggered the action.

        :return: A tuple containing:
                 whether the event request was sent successfully; and
                 the name of the event that was sent.
        """
        response = requests.get(self._URL.format(
            self._webhooks_event_name,
            self._webhooks_key))

        return response.ok, self._webhooks_event_name

Action #2

Notify a neighbour via SMS.

Below is a custom action, BoboActionTwilioSMS, that uses the Twilio SMS API to accomplish this. Note: the code below requires the additional twilio package.

Code

from typing import Tuple, Any
from bobocep.cep.action import BoboAction
from bobocep.cep.event import BoboEventComplex
from twilio.rest import Client  # https://pypi.org/project/twilio/


class BoboActionTwilioSMS(BoboAction):
    """
    An action that sends an SMS via the Twilio API.
    """

    def __init__(
            self,
            name: str,
            account_sid: str,
            auth_token: str,
            num_from: str,
            num_to: str,
            message: str,
            *args,
            **kwargs):
        """
        :param name: The action name.
        :param account_sid: Twilio Account SID.
        :param auth_token: Twilio Auth Token.
        :param num_from: Twilio phone number.
        :param num_to: Recipient phone number.
        :param message: Message to send to recipient.
        :param args: Action arguments.
        :param kwargs: Action keyword arguments.
        """
        super().__init__(name=name, args=args, kwargs=kwargs)

        self._client = Client(account_sid, auth_token)
        self._num_from = num_from
        self._num_to = num_to
        self._message = message

    def execute(self, event: BoboEventComplex) -> Tuple[bool, Any]:
        """
        :param event: The complex event that triggered the action.

        :return: A tuple containing:
                 whether the SMS was sent successfully; and
                 the recipient phone number.
        """
        message = self._client.messages.create(
            from_=self._num_from,
            body=self._message,
            to=self._num_to
        )

        success = message.status in ("delivered", "queued", "sending", "sent")
        return success, self._num_to

Or, notify multiple neighbours with a sequential action.

Each action in actions below would be an instance of BoboActionTwilioSMS but with differing num_to values, which represents the recipient’s phone number.

Code

action_notify_neighbours = BoboActionMultiSequential(
    name="Notify Neighbours",
    actions=[action_neighbour_1, action_neighbour_2],
    stop_on_fail=False
)