in src/webcam_client.py [0:0]
def detect_ppe(frame, lambda_client):
# Resize frame to 1/2 for faster processing
small = cv2.resize(frame, (0, 0), fx=0.5, fy=0.5)
# Encode to JPG and send to lambda
ret, encoded = cv2.imencode('.jpg', small)
if not ret:
raise RuntimeError('Failed to encode frame')
account_id = boto3.client('sts').get_caller_identity().get('Account')
response = lambda_client.invoke(
FunctionName=f'ppe-detection-function-{account_id}',
InvocationType='RequestResponse',
Payload=json.dumps({
'camera_ID': CAMERA_ID,
'alarm_ID': ALARM_ID,
'img': base64.b64encode(encoded).decode('utf-8')
}))
# Annotate bounding boxes to frame
response_dict = json.loads(response['Payload'].read())
if 'FunctionError' not in response:
annotate_frame(frame, response_dict)
else:
print(response_dict['errorMessage'])