Send WeChat Notifications
Introduction¶
In this tutorial, we will walk you through a simple use case for BrainFrame: getting WeChat Notification when there is no cashier in the checkout area. You can find the complete script on our GitHub repository.
Setup The Environment¶
In a previous tutorial, we installed the BrainFrame server, client, and Python API libraries. In this tutorial, the API functions we are going to use are:
api.set_stream_configuration(...)
api.set_zone(...)
api.get_latest_zone_statuses()
api.get_zone_status_stream()
We will be using a third-party library called itchat to send
notifications to WeChat. We'll install it using pip
:
pip3 install itchat
We will also use one of our publicly available capsules,
detector_people_and_vehicles_fast
. You can grab it from our
downloads page.
Before we start, you should have the BrainFrame server and client running, and capsules ready.
Log In to WeChat¶
As usual, we will begin by importing our dependencies:
from pathlib import Path
import itchat as wechat
from brainframe.api import BrainFrameAPI, bf_codecs
Then, let's log in to our WeChat account and send a test message:
wechat.auto_login()
wechat.send_msg(f"Notifications from BrainFrame have been enabled",
toUserName="filehelper")
The script will display a QR code, Scan it with your WeChat app and login. Your File Helper will then receive the message.
Create a New Stream from a Local File¶
First set the BrainFrame URL:
api = BrainFrameAPI("http://localhost")
We will reuse the code snippet introduced in the previous tutorial to create a stream configuration on the BrainFrame server. We're going to use a simulated video file for this demo, but it will work with live video streams as well.
# Upload the local file to the BrainFrame server's database and get its storage
# ID
storage_id = api.new_storage(
data=Path("../videos/shopping_cashier_gone.mp4").read_bytes(),
mime_type="application/octet-stream"
)
# Create a StreamConfiguration with the storage ID
new_stream_config = bf_codecs.StreamConfiguration(
# The display name on the client side
name="Demo",
# Specify that we're using a file
connection_type=bf_codecs.StreamConfiguration.ConnType.FILE,
connection_options={
# The storage id of the file
"storage_id": storage_id,
},
runtime_options={},
premises_id=None,
)
# Send the StreamConfiguration to the server to have it connect
new_stream_config = api.set_stream_configuration(new_stream_config)
# Tell the server to start analysis on the new stream
api.start_analyzing(new_stream_config.id)
You can download the demo video from our tutorial scripts repository. We recorded a video simulating a cashier serving customers.
Create a Zone and Setup an Alarm¶
In BrainFrame, alarms are associated with zones, and you can configure them through the client or through the API. You can check our documentation on Zones and Alarms for more information.
Using the API, we will create a zone around the check-out counter, and an alarm that will be triggered if no people are in that zone.
# Condition for the Alarm that will trigger when there is <1 person in the zone
# that it is assigned to
no_cashier_alarm_condition = bf_codecs.ZoneAlarmCountCondition(
test=bf_codecs.CountConditionTestType.LESS_THAN,
check_value=1,
with_class_name="person",
with_attribute=None,
window_duration=5.0,
window_threshold=0.5,
intersection_point=bf_codecs.IntersectionPointType.BOTTOM,
)
# Create the ZoneAlarm. It will be active all day, everyday and will be
# triggered if the detection results satisfy the condition we created. Because
# use_active_time==False, the active end/start times will be ignored.
no_cashier_alarm = bf_codecs.ZoneAlarm(
name="Missing Cashier!",
count_conditions=[no_cashier_alarm_condition],
rate_conditions=[],
use_active_time=False,
active_start_time="00:00:00",
active_end_time="23:59:59",
)
# Create a Zone object with the above alarm
cashier_zone = bf_codecs.Zone(
name="Cashier",
stream_id=new_stream_config.id,
alarms=[no_cashier_alarm],
coords=[[513, 695], [223, 659], [265, 340], [513, 280], [578, 462]]
)
# Send the Zone to BrainFrame
api.set_zone(cashier_zone)
In the client, you will be able to see the zone there:
Get Zone Status¶
In BrainFrame, we use the ZoneStatus
data structure to represent the inference
results of frames. Let's use it to get ours.
We can use the API to get the latest ZoneStatus
objects from BrainFrame.
zone_statuses = api.get_latest_zone_statuses()
print("Zone Statuses: ", zone_statuses)
The above code will print out the latest ZoneStatus
objects for each stream
with analysis/inference enabled. Warning: it can be a very long data structure,
depending on how many streams there are and what capsules are loaded.
This is the most direct way to get the most recent inference results from BrainFrame. However, you have to call this function each time you want new results, which is a hassle.
A different API function, get_zone_status_stream()
helps alleviate this issue.
Instead of having relying on you polling for ZoneStatus
objects, this function
will return an iterable object to you. Each time BrainFrame has a new result
available, it will be pushed to the iterator.
zone_status_iterator = api.get_zone_status_stream()
for zone_statuses in zone_status_iterator:
print("Zone Statuses: ", zone_statuses)
This script will print the zone statuses as fast as the capsules can process the frames.
Get Alarms and Send Notifications to WeChat¶
We can iterate through the zone status packets and check if there are any alerts that recently terminated after lasting >5 seconds. If there were, we send a notification. Note that for this example, the alert will only trigger after the cashier returns to the counter, a situation that is not as useful outside of the demo environment. The script will also only send one notification before exiting, to avoid sending too many notifications.
# Iterate through the zone status packets
for zone_status_packet in zone_status_iterator:
for stream_id, zone_statuses in zone_status_packet.items():
for zone_name, zone_status in zone_statuses.items():
for alert in zone_status.alerts:
# Check if the alert has ended
if alert.end_time is None:
continue
total_time = alert.end_time - alert.start_time
# Check if the alert lasted for more than 5 seconds
if total_time > 5:
alarm = api.get_zone_alarm(alert.alarm_id)
wechat.send_msg(
f"BrainFrame Alert: {alarm.name} \n"
f"Duration {total_time}", toUserName="filehelper")
# Stop here, for demo purposes
exit()
The script will send an alert to your WeChat File Helper if the cashier has been missing for more than 5 seconds. It will then exit the loop.
Logout Your WeChat Account¶
Finally, before we exit the script, don't forget to log out your WeChat account.
Put the following code above exit()
.
wechat.logout()