-
Notifications
You must be signed in to change notification settings - Fork 0
/
lambda_function.py
241 lines (203 loc) · 12.1 KB
/
lambda_function.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
import json
import logging
import csv
from PIL import Image
import io
import boto3
import email
import requests
from twilio.rest import Client
from fuzzywuzzy import fuzz
import datetime
import time
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import base64
# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Initialize the S3 client
s3_client = boto3.client('s3')
# Initialize the SES client
ses_client = boto3.client('ses')
# Initialize the SSM client
ssm_client = boto3.client('ssm')
# Retrieve sensitive data and configuration parameters from Parameter Store
def get_parameter(parameter_name):
try:
response = ssm_client.get_parameter(Name=parameter_name, WithDecryption=True)
return response['Parameter']['Value']
except Exception as e:
logger.error(f'Error retrieving parameter {parameter_name}: {str(e)}')
return None
# Retrieve sensitive data and configuration parameters from Parameter Store
plate_recognizer_token = get_parameter('/LicensePlateRecognition/PlateRecognizerToken')
ses_sender_email = get_parameter('/LicensePlateRecognition/SESSenderEmail')
ses_email_notification_to = get_parameter('/LicensePlateRecognition/SESEmailNotificationTo')
twilio_account_sid = get_parameter('/LicensePlateRecognition/TwilioAccountSID')
twilio_auth_token = get_parameter('/LicensePlateRecognition/TwilioAuthToken')
twilio_from_phone_number = get_parameter('/LicensePlateRecognition/TwilioFromPhoneNumber')
twilio_to_phone_number = get_parameter('/LicensePlateRecognition/TwilioToPhoneNumber') # Retrieve Twilio To phone number from SSM
s3_bucket_name = get_parameter('/LicensePlateRecognition/S3BucketName')
s3_file_key = get_parameter('/LicensePlateRecognition/S3FileKey')
fuzzy_match_threshold = int(get_parameter('/LicensePlateRecognition/FuzzyMatchThreshold'))
raw_inbound_email_bucket = get_parameter('/LicensePlateRecognition/RawInboundEmailBucket') # Ensure this parameter exists in SSM
# Send an email notification using SES with execution time and a smaller attachment
def send_email_notification(recipient, subject, message_body, script_start_time):
try:
# Include explanatory text, script start time, and elapsed time in the email body
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
elapsed_time = time.time() - script_start_time # Calculate the elapsed time
elapsed_time_formatted = f'{elapsed_time:.1f}' # Format elapsed time to one decimal place
# Convert the script_start_time timestamp to a datetime object
script_start_time_datetime = datetime.datetime.fromtimestamp(script_start_time)
# Format the datetime object as a string
formatted_script_start_time = script_start_time_datetime.strftime("%Y-%m-%d %H:%M:%S")
message_body_with_time = (
f'### Script Start Time: {formatted_script_start_time} ###\n\n'
f'{message_body}\n\n'
f'Current Time: {current_time}\n'
f'Elapsed Time: {elapsed_time_formatted} seconds'
)
# Create a MIME email message with attachments
msg = MIMEMultipart()
msg['From'] = ses_sender_email
msg['To'] = recipient
msg['Subject'] = subject
# Attach the message body
msg.attach(MIMEText(message_body_with_time, 'plain'))
# Open and resize the image
with open('/tmp/attachment.jpg', 'rb') as attachment_file:
img = Image.open(attachment_file)
img.thumbnail((600, 600)) # Resize the image to fit within a 300x300 pixel box
img_byte_array = io.BytesIO()
img.save(img_byte_array, format='JPEG')
img_data = img_byte_array.getvalue()
# Attach the resized image
attachment = MIMEBase('application', 'octet-stream')
attachment.set_payload(img_data)
encoders.encode_base64(attachment)
attachment.add_header('Content-Disposition', f'attachment; filename=attachment.jpg')
msg.attach(attachment)
# Send the email
raw_message = msg.as_string()
response = ses_client.send_raw_email(
Source=ses_sender_email,
Destinations=[recipient],
RawMessage={'Data': raw_message}
)
logger.info(f'SES Email sent with execution time and attachment: {response["MessageId"]}')
except Exception as e:
logger.error(f'Error sending SES email: {str(e)}')
# Make a Twilio call to open the gate
def make_twilio_call(registered_to):
try:
twilio_to_number = twilio_to_phone_number # Retrieve Twilio To phone number from SSM
twilio_from_number = twilio_from_phone_number
# Customize your call message
call_message = f"Match found! This vehicle is registered to {registered_to}."
# Make the Twilio call
twilio_client = Client(twilio_account_sid, twilio_auth_token)
call = twilio_client.calls.create(
url="http://demo.twilio.com/docs/voice.xml",
to=twilio_to_number,
from_=twilio_from_number,
twiml=f'<Response><Say>{call_message}</Say></Response>'
)
logger.info(f'Twilio call SID: {call.sid}')
except Exception as e:
logger.error(f'Error making Twilio call: {str(e)}')
# Lambda handler function
def lambda_handler(event, context):
script_start_time = time.time() # Record the start time
try:
for record in event.get('Records', []):
# Extract email information from the event record
ses_event = record.get('ses', {}).get('mail', {})
message_id = ses_event.get('messageId')
# Log the event for inspection
logger.info(f'Email Message Id: {message_id}')
# Download the email content from the source S3 bucket
s3_client.download_file(raw_inbound_email_bucket, message_id, '/tmp/email.eml')
# Parse the email and find attachments
with open('/tmp/email.eml', 'rb') as email_file:
msg = email.message_from_binary_file(email_file)
for part in msg.walk():
if part.get_content_maintype() == 'multipart':
continue
filename = part.get_filename()
if filename:
# Save the attachment to a temporary file
with open('/tmp/attachment.jpg', 'wb') as attachment_file:
attachment_file.write(part.get_payload(decode=True))
# Upload the attachment to the Plate Recognizer API using requests
regions = ["ie"] # Change to your country
with open('/tmp/attachment.jpg', 'rb') as fp:
response = requests.post(
'https://api.platerecognizer.com/v1/plate-reader/',
data=dict(regions=regions), # Optional
files=dict(upload=fp),
headers={'Authorization': f'Token {plate_recognizer_token}'})
# Log the response from Plate Recognizer API
response_text = response.text
logger.info(f'Plate Recognizer API Response Status Code: {response.status_code}')
logger.info(f'Plate Recognizer API Response: {response_text}')
# Extract the recognized plate value from the API response
plate_recognized = None
try:
response_data = json.loads(response_text)
if 'results' in response_data:
first_result = response_data['results'][0]
plate_recognized = first_result.get('plate', '').lower()
if 'score' in first_result:
score = float(first_result.get('score'))
else:
score = 0.0 # Default score if not found
except Exception as e:
logger.error(f'Error extracting recognized plate: {str(e)}')
# Now, retrieve the CSV file from S3 and store its contents in a dictionary
csv_content = s3_client.get_object(Bucket=s3_bucket_name, Key=s3_file_key)['Body'].read().decode('utf-8')
# Initialize an empty dictionary to store the CSV data
csv_data = {}
# Parse the CSV content line by line
csv_reader = csv.reader(io.StringIO(csv_content))
for row in csv_reader:
if len(row) >= 2:
key, value = row[0].strip(), row[1].strip()
csv_data[key.lower()] = value.lower()
# Log the results
logger.info('CSV data for authorised licence plate numbers:')
logger.info(csv_data)
# Compare the recognized plate to the values in the CSV (including fuzzy matching)
best_match = None
for csv_key in csv_data.keys():
match_score = fuzz.partial_ratio(plate_recognized, csv_key)
if match_score >= fuzzy_match_threshold:
best_match = csv_key
break
if best_match is not None:
matched_value = csv_data.get(best_match, '')
logger.info(f'Match found for vehicle license plate number: {plate_recognized}, Registered to: {matched_value}')
# Send an email notification when a match is found
if score == 1.0:
make_twilio_call(matched_value) # Make a Twilio call for both exact and fuzzy matches to open the gate
send_email_notification(ses_email_notification_to, f'Gate Opening Alert - Exact Match Found for Plate: {plate_recognized}',
f'Exact match, licence plate number {plate_recognized} is registered to: {matched_value}', script_start_time)
else:
make_twilio_call(matched_value) # Make a Twilio call for both exact and fuzzy matches to open the gate
send_email_notification(ses_email_notification_to, f'Gate Opening Alert - Fuzzy Match Found for Plate: {plate_recognized}',
f'Fuzzy match, licence plate number {plate_recognized}is registered to: {matched_value}', script_start_time)
else:
logger.info(f'No match found for vehicle license plate number: {plate_recognized}')
# Send an email notification when no match is found for debugging, logging, and alerting purposes
send_email_notification(ses_email_notification_to, f'Failed Gate Opening Alert - No Match Found for Plate: {plate_recognized}',
f'No match found or vehicle not registered for licence plate number: {plate_recognized}', script_start_time)
except Exception as e:
# Log the error message
logger.error(f'Error: {str(e)}')
return {
'statusCode': 200,
'body': json.dumps('Email processing complete.')
}