Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Python script for server side of IoT BOT to monitor BLE tags (for AWS lambda)
# -*- coding: utf-8 -*-
from datetime import datetime, timedelta, timezone
import boto3
from boto3.session import Session
from boto3.dynamodb.conditions import Key
TIMEOUT_SEC = 180
def lambda_handler(event, context):
# 現在時刻を取得
JST = timezone(timedelta(hours=+9), 'JST')
now = datetime.now(JST)
now_ut = int(now.strftime('%s'))
now_dt = now.strftime('%Y/%m/%d %H:%M:%S (JST)')
try:
# dynamodbのテーブルに接続
region = "ap-northeast-1"
session = Session(
region_name = region
)
dynamodb = session.resource('dynamodb')
table = dynamodb.Table('{your_table_name}')
response = {}
# 入力がタグ情報の場合、テーブルにタグ情報を追加(状態は不在=0)
if 'tag_info' in event.keys():
response['inserted'] = []
for info in event['tag_info']:
tid = info['tid']
name = info['name']
table.put_item(
Item = {
'tid' : tid,
'name' : name,
'exists' : 0,
'utime' : now_ut,
'datetime' : now_dt
}
)
response['inserted'].append({'tid': tid, 'name': name})
# 入力がタグ状態の場合、タグ状態に基づいてテーブルを更新しアクションを決定
# アクション:1=出現,0=消失
if 'tag_detected' in event.keys():
response['updated'] = []
queryData = table.scan()
for data in queryData['Items']:
tid = data['tid']
name = data['name']
exists = data['exists']
utime = data['utime']
dtime = data['datetime']
if tid in event['tag_detected']:
table.update_item(
Key = {
'tid' : tid
},
UpdateExpression = "set #e=:e, utime=:ut, #d=:dt",
ExpressionAttributeNames = { "#e":"exists", "#d":"datetime" },
ExpressionAttributeValues = {
':e' : 1,
':ut' : now_ut,
':dt' : now_dt
}
)
if not exists:
response['updated'].append({
'name': name,
'action': 1
})
elif exists and now_ut > utime + TIMEOUT_SEC:
table.update_item(
Key = {
'tid' : tid
},
UpdateExpression = "set #e=:e",
ExpressionAttributeNames = { "#e": "exists" },
ExpressionAttributeValues = {
':e' : 0
}
)
response['updated'].append({
'name': name,
'action': 0
})
# モニターからの受信ログを記録
if 'monitor_name' in event.keys():
table_log = dynamodb.Table('BLETagMonitor_MonitorLog')
table_log.put_item(
Item = {
'name' : event['monitor_name'],
'utime' : now_ut,
'datetime' : now_dt
}
)
return response
except Exception as e:
print(e)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.