Device #{device_id}
{location_combobox}
{description_textbox}
"""
def populate_welcome_email(template_content, user_data):
"""
Populates the welcome email template with user-specific data.
Args:
template_path (str): The file path to the HTML email template.
user_data (dict): A dictionary containing user information:
'first_name', 'last_name', 'user_name',
'password', 'devices' (a list of device IDs).
Returns:
str: The populated HTML content of the email.
"""
# Use str.replace() for each placeholder to avoid conflicts with CSS/JS
content = template_content.replace('{first_name}', user_data.get("first_name", ""))
content = content.replace('{last_name}', user_data.get("last_name", ""))
content = content.replace('{user_name}', user_data.get("user_name", ""))
content = content.replace('{password}', user_data.get("password", ""))
content = content.replace('{signature}', user_data.get("signature", ""))
return content
def read_file(file_name, source = "LOCAL", type_ = "TEXT", bucket_name="daily-maps"):
blob_data = ""
if source == "MINIO":
#blob_data = ReadObjectMinIO(bucket_name, file_name)
pass
elif source == "LOCAL":
login_file = os.path.join(filesDir, file_name)
login_file = login_file.replace("\\","/")
logger.debug(f"Full file path: {login_file}")
logger.debug(f"File exists: {os.path.exists(login_file)}")
#print(login_file)
if type_ == "TEXT":
with open(login_file, encoding="utf8") as f:
blob_data = f.read()
else:
with open(login_file, 'rb') as f:
blob_data = f.read()
elif source == "AZURE":
try:
blob_data = ""#container_client.download_blob(file_name).readall()
except Exception as err:
logger.error("Not reading Azure blob "+str(err))
blob_data = ""
return blob_data
else:
pass
return blob_data
# --- Main Monitoring Loop ---
def monitor_devices():
"""Main monitoring loop that checks devices periodically."""
logger.info("Starting monitoring loop...")
while not stop_event.is_set():
try: # Add this outer try block
start_time = time.monotonic()
#current_time_unix = time.time()
#======================== Do node-red instances management ==================================================
hash_data = GetRedisMap('node-red_deployed')
#logger.debug(f"node_red_requests: {hash_data}")
requests_count = 0
if hash_data != {}:
requests_count = int(hash_data['requests'])
if requests_count > 0:
logger.debug(f"node-red_deployed: {str(hash_data)}")
user_name = hash_data['user_name']
#delete request #this might need switching to queue... todo
redis_conn.hset('node-red_deployed', mapping={
'user_name': user_name,
'token': "",
'time': "",
'requests': 0
})
sync_success = False
changes_were_made = False # Flag to track if sync modified files
signal_file = Path("/tmp/node-red-sync-changes.signal") # Define path again
try:
# Use sys.executable to ensure using the same python interpreter
print("Starting node synchronization...")
sync_script_path = "/home/ubuntu/.node-red/sync_nodes.py"
# Add error checking
result = subprocess.run(
[sys.executable, sync_script_path],
check=True, # Raise CalledProcessError if script exits with non-zero code
capture_output=True,
text=True,
timeout=600 # Add a generous timeout (10 minutes?) for npm operations
)
print("Node synchronization script output:\n", result.stdout)
if result.stderr: print("Node synchronization script stderr:\n", result.stderr)
print("Node synchronization completed successfully.")
sync_success = True
if signal_file.exists():
changes_were_made = True
logger.info("Sync script indicated changes were made (signal file found).")
try:
signal_file.unlink() # Consume the signal
logger.debug("Removed signal file.")
except OSError as e:
logger.warning(f"Could not remove signal file {signal_file}: {e}")
else:
logger.info("Sync script indicated no changes were made (signal file not found).")
print("No node changes detected by sync script.")
except FileNotFoundError:
print(f"ERROR: Sync script not found at {sync_script_path}")
# Decide how to handle: exit, skip deployment, etc.
#return # Or raise an exception
except subprocess.CalledProcessError as e:
print(f"ERROR: Node synchronization script failed with exit code {e.returncode}")
print(f"Stderr:\n{e.stderr}")
print(f"Stdout:\n{e.stdout}")
# Decide how to handle: exit, skip deployment, etc.
#return # Or raise an exception
except subprocess.TimeoutExpired:
print(f"ERROR: Node synchronization script timed out.")
# Decide how to handle
#return # Or raise an exception
except Exception as e_sync:
print(f"ERROR: An unexpected error occurred while running sync script: {e_sync}")
#return # Or raise
# --- 2. Restart Node-RED Service IF Sync Succeeded AND Changes Were Made ---
restart_successful = True # Default to true if no restart needed
if sync_success and changes_were_made: # Only restart if sync OK AND changes happened
print("Node changes detected, restarting Node-RED service...")
restart_successful = restart_node_red_service("node-red.service")
if not restart_successful:
print("ERROR: Node-RED service failed to restart properly after changes. Aborting deployment.")
return # Exit processing this request
elif sync_success and not changes_were_made:
print("No node changes required restart.")
else:
# Sync failed, don't restart
print("Skipping Node-RED restart due to sync script failure.")
# --- 3. Combine and Deploy Flows (only if sync didn't fail AND restart was ok/not needed) ---
if sync_success and restart_successful: # Proceed if sync ran ok and restart (if needed) was ok
print("Proceeding with flow combination and deployment...")
# ... (rest of combine and deploy logic) ...
else:
print("Skipping flow deployment due to previous errors during sync or restart.")
# Get list of user directories
base_dir = "/home/ubuntu/.node-red/users"
user_dirs = [os.path.join(base_dir, d) for d in os.listdir(base_dir)
if os.path.isdir(os.path.join(base_dir, d))]
# Combine flows
output_dir = "/home/ubuntu/.node-red/main_instance"
flow_file = combine_user_flows(user_dirs, output_dir)
# Deploy combined flow
creds_file = os.path.join(output_dir, "common_flow_cred.json")
success = deploy_combined_flow(
flow_file,
node_red_url="https://eluxnetworks.net:1999",
credentials_file=creds_file,
admin_user=ADMIN_USER,
admin_password=ADMIN_PASSWORD
)
#print(f"Combined flow created at: {combined_file}")
#lets check if any node-reds are stale
pattern = 'node_red_status_*'
threshold = time.time() - 10
# Find all matching keys
matching_keys = []
cursor = 0
while True:
cursor, keys = redis_conn.scan(cursor=cursor, match=pattern, count=100)
for key in keys:
key_str = key.decode('utf-8') # Convert bytes to string
# Get the last_activity value
last_activity = redis_conn.hget(key_str, 'last_activity').decode('utf-8')
pid = 0
try:
pid = int(redis_conn.hget(key_str, 'pid'))
if pid > 0:
if last_activity:
last_activity_value = int(float(last_activity))
# Check if it's below the threshold
print(last_activity_value-threshold)
if last_activity_value < threshold:
#if True:
if key_str not in permament_users:
matching_keys.append(key_str)
except:
pass
# Exit the loop when scan is complete
if cursor == 0:
break
# Print the result
if len(matching_keys) > 0:
print(f"Found {len(matching_keys)} keys with last_activity < {threshold}:")
for key in matching_keys:
print(key)
user_name = key[len("node_red_status_"):]
time_s = time.time()#datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
port, pid = GetNodeRedDetails(user_name)
result = StopNodeRed(pid)
#Lets store it to postgres and REDIS
StoreNodeRedUserPortOff(port, user_name, time_s)
redis_conn.hset(f'node_red_status_{user_name}', mapping={
'port': 0,
'started_time': time_s,
'last_activity': time_s,
'pid': 0,
})
print(result)
# --- Cycle Cleanup & Wait ---
elapsed_time = time.monotonic() - start_time
#logger.info(f"Monitoring cycle finished in {elapsed_time:.2f} seconds.")
wait_time = CHECK_INTERVAL_SECONDS - elapsed_time
if wait_time > 0:
#logger.debug(f"Waiting for {wait_time:.2f} seconds before next cycle.")
# Use stop_event.wait() for graceful shutdown during sleep
stop_event.wait(wait_time)
else:
logger.warning("Monitoring cycle took longer than the configured interval.")
except Exception as e:
logger.exception(f"Critical error in monitor_devices outer loop: {e}")
time.sleep(10) # Wait before retrying the whole loop
logger.info("Monitoring loop stopped.")
def monitor_messages():
"""Main monitoring loop that checks devices periodically."""
logger.info("Starting monitoring loop...")
last_id = '0'
try:
while not stop_event.is_set():
try: # Add this outer try block
#======================== Process messaging requests ========================================================
# Read from stream
#logger.debug("About to read from Redis stream...")
messages = redis_conn.xread({'messaging_requests_stream': last_id})
#logger.debug(f"Redis read successful, got {len(messages) if messages else 0} messages")
#print(".", end="")
if messages:
call_acl = False
for stream, msgs in messages:
for msg_id, fields in msgs:
try:
logger.debug(f"Processing messaging request: {str(fields)}")
function = fields[b'function'].decode('utf-8')
if function == "new_caretaker":
email = fields[b'email'].decode('utf-8')
user_name = fields[b'user_name'].decode('utf-8')
first_name = fields[b'first_name'].decode('utf-8')
last_name = fields[b'last_name'].decode('utf-8')
phone_number = fields[b'phone_number'].decode('utf-8')
password = fields[b'password'].decode('utf-8')
#devices = fields[b'devices'].decode('utf-8')
signature = fields[b'signature'].decode('utf-8')
#user_name = hash_data['user_name']
#first_name = hash_data['first_name']
#last_name = hash_data['last_name']
#phone_number = hash_data['phone_number']
#password = hash_data['password']
#devices = hash_data['devices']
#signature = hash_data['signature']
#requests = hash_data['requests']
template_html = read_file("welcome_template_short.html")
#devices_lst = devices.split(",")
new_user = {
"first_name": first_name,
"last_name": last_name,
"user_name": user_name,
"phone_number": phone_number,
"signature": signature,
"password": password
}
email_body = populate_welcome_email(template_html, new_user)
# 4. Save the populated email to a new HTML file to preview it
if "Error" not in email_body:
with open('populated_welcome_email.html', 'w') as file:
file.write(email_body)
print("Successfully generated 'populated_welcome_email.html'.")
# In a real application, you would use a library like smtplib to send this 'email_body'
text_str = "CREDS," + first_name + "," + last_name + "," + user_name + "," + password + "," + email
#SendMessageTo(user_id, text_str)
#need to send it to device via signature, since, if device is new, not MQTT_id is known.
topic = "/well_" + signature
SendMessageDirectTo(topic, text_str)
sender.SendEmailTo(email, email_body, "Wellcome to WellNuo!")
call_acl = True
else:
print(email_body)
redis_conn.xdel('messaging_requests_stream', msg_id) #uncomment later
last_id = msg_id
elif function == "new_beneficiary":
email = fields[b'email'].decode('utf-8')
user_name = fields[b'user_name'].decode('utf-8')
first_name = fields[b'first_name'].decode('utf-8')
last_name = fields[b'last_name'].decode('utf-8')
phone_number = fields[b'phone_number'].decode('utf-8')
password = fields[b'password'].decode('utf-8')
#devices = fields[b'devices'].decode('utf-8')
signature = fields[b'signature'].decode('utf-8')
#user_name = hash_data['user_name']
#first_name = hash_data['first_name']
#last_name = hash_data['last_name']
#phone_number = hash_data['phone_number']
#password = hash_data['password']
#devices = hash_data['devices']
#signature = hash_data['signature']
#requests = hash_data['requests']
template_html = read_file("new_beneficiary_template_short.html")
#devices_lst = devices.split(",")
new_user = {
"first_name": first_name,
"last_name": last_name,
"user_name": user_name,
"phone_number": phone_number,
"signature": signature,
"password": password
}
email_body = populate_welcome_email(template_html, new_user)
# 4. Save the populated email to a new HTML file to preview it
if "Error" not in email_body:
with open('populated_welcome_email.html', 'w') as file:
file.write(email_body)
print("Successfully generated 'populated_welcome_email.html'.")
# In a real application, you would use a library like smtplib to send this 'email_body'
text_str = "CREDS," + first_name + "," + last_name + "," + user_name + "," + password + "," + email
#SendMessageTo(user_id, text_str)
#need to send it to device via signature, since, if device is new, not MQTT_id is known.
#lets NOT send message here... because adding new beneficiary/deployment would cause credentials on caretaker's phone to change
#topic = "/well_" + signature
#SendMessageDirectTo(topic, text_str)
sender.SendEmailTo(email, email_body, "Wellcome to WellNuo!")
call_acl = True
else:
print(email_body)
redis_conn.xdel('messaging_requests_stream', msg_id) #uncomment later
last_id = msg_id
elif function == "credentials_updated":
email = fields[b'email'].decode('utf-8')
user_name = fields[b'user_name'].decode('utf-8')
first_name = fields[b'first_name'].decode('utf-8')
last_name = fields[b'last_name'].decode('utf-8')
phone_number = fields[b'phone_number'].decode('utf-8')
password = fields[b'password'].decode('utf-8')
#devices = fields[b'devices'].decode('utf-8')
signature = fields[b'signature'].decode('utf-8')
#user_name = hash_data['user_name']
#first_name = hash_data['first_name']
#last_name = hash_data['last_name']
#phone_number = hash_data['phone_number']
#password = hash_data['password']
#devices = hash_data['devices']
#signature = hash_data['signature']
#requests = hash_data['requests']
template_html = read_file("new_credentials_template_short.html")
#devices_lst = devices.split(",")
new_user = {
"first_name": first_name,
"last_name": last_name,
"user_name": user_name,
"phone_number": phone_number,
"signature": signature,
"password": password
}
email_body = populate_welcome_email(template_html, new_user)
# 4. Save the populated email to a new HTML file to preview it
if "Error" not in email_body:
with open('populated_welcome_email.html', 'w') as file:
file.write(email_body)
print("Successfully generated 'populated_welcome_email.html'.")
# In a real application, you would use a library like smtplib to send this 'email_body'
text_str = "CREDS," + first_name + "," + last_name + "," + user_name + "," + password + "," + email
#SendMessageTo(user_id, text_str)
#need to send it to device via signature, since, if device is new, not MQTT_id is known.
topic = "/well_" + signature
SendMessageDirectTo(topic, text_str)
sender.SendEmailTo(email, email_body, "Wellcome to WellNuo!")
call_acl = True
else:
print(email_body)
redis_conn.xdel('messaging_requests_stream', msg_id) #uncomment later
last_id = msg_id
elif function == "update_acl":
call_acl = True
redis_conn.xdel('messaging_requests_stream', msg_id) #uncomment later
last_id = msg_id
except Exception as e:
print(f"Error in Processing messaging request: {e}")
#call acl_manager.py from /home/ubuntu/mqtt-auth-service. New user was added in well-api
if call_acl:
try:
result = subprocess.run(
[sys.executable, '/home/ubuntu/mqtt-auth-service/acl_manager.py'],
capture_output=True,
text=True,
timeout=60 # Add timeout to prevent hanging
)
if result.returncode == 0:
print("ACL manager executed successfully")
print("Output:", result.stdout)
else:
print("ACL manager failed with return code:", result.returncode)
print("Error:", result.stderr)
except subprocess.TimeoutExpired:
print("ACL manager execution timed out")
except Exception as e:
print(f"Error executing ACL manager: {e}")
time.sleep(0.1)
hash_data = GetRedisMap('node_red_requests')
#logger.debug(f"node_red_requests: {hash_data}")
requests_count = 0
if hash_data != {}:
requests_count = int(hash_data['requests'])
if requests_count > 0:
logger.debug(f"node_red_requests: {str(hash_data)}")
user_name = hash_data['user_name']
#user needs Node-red. Is his session up and running?
port, pid = IsNRRunning(user_name)
#delete request #this might need switching to queue... todo
redis_conn.hset('node_red_requests', mapping={
'user_name': user_name,
'token': "",
'time': "",
'requests': 0
})
if port == 0 or pid == 0: #not running
token = hash_data['token']
time_s = time.time()#datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
port = GetAvailablePort(user_name)
users_p = GetUserPriviledges(user_name)
ps = users_p[0][1]
SetupUserEnvironment(user_name, token, ps)
pid = StartNodeRed(port, user_name)
#Lets store it to postgres and REDIS
StoreNodeRedUserPortOn(port, user_name, pid, time_s)
redis_conn.hset(f'node_red_status_{user_name}', mapping={
'port': port,
'started_time': time_s,
'last_activity': time_s,
'pid': pid,
})
else:
#time_s = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
time_s = time.time()
redis_conn.hset(f'node_red_status_{user_name}', mapping={
'port': port,
'started_time': time_s,
'last_activity': time.time(),
'pid': pid
})
except redis.RedisError as e:
logger.error(f"Redis error in message monitoring: {e}")
time.sleep(5) # Wait before retry
except Exception as e:
logger.error(f"Unexpected error in message processing: {e}")
time.sleep(1)
except Exception as e:
logger.exception(f"Critical error in monitor_messages: {e}")
finally:
logger.info("monitor_messages loop stopped.")
# --- Signal Handling ---
def signal_handler(signum, frame):
"""Handles termination signals gracefully."""
logger.warning(f"Received signal {signal.Signals(signum).name}. Initiating graceful shutdown...")
stop_event.set()
t1.stop()
t2.stop()
for job_id in job_registry:
function_name = job_registry[job_id]['function_name']
argument = job_registry[job_id]['argument']
task_key = f"{function_name}_{argument}"
# Remove from task registry if this is the current job for this task/arg
if task_key in task_arg_registry and task_arg_registry[task_key] == job_id:
del task_arg_registry[task_key]
# Remove the job from the scheduler
scheduler.remove_job(job_id)
logger.info(f"Cancelled job '{job_id}' with argument '{job_registry[job_id]['argument']}'")
def FahrenheitToCelsius(F):
if isinstance(F, str):
F = float(F)
C = (F - 32) * 5/9
return C
def BitIndex(bit_nr):
return -bit_nr - 1
def RadarLarger(radar, radar_threshold):
thr_parts = radar_threshold[0].split("_")
field = thr_parts[0]
if field == "s28":
radar_val = sum(radar[1][3:]) / (7 * radar[1][0])
elif field == "m0":
radar_val = radar[0][5] / radar[1][0]
elif field == "m1":
radar_val = radar[0][6] / radar[1][0]
elif field == "m2":
radar_val = radar[0][7] / radar[1][0]
elif field == "m3":
radar_val = radar[0][8] / radar[1][0]
elif field == "m4":
radar_val = radar[0][9] / radar[1][0]
elif field == "m5":
radar_val = radar[0][10] / radar[1][0]
elif field == "m6":
radar_val = radar[0][11] / radar[1][0]
elif field == "m7":
radar_val = radar[0][12] / radar[1][0]
elif field == "m8":
radar_val = radar[0][13] / radar[1][0]
elif field == "s2":
radar_val = radar[1][3] / radar[1][0]
elif field == "s3":
radar_val = radar[1][4] / radar[1][0]
elif field == "s4":
radar_val = radar[1][5] / radar[1][0]
elif field == "s5":
radar_val = radar[1][6] / radar[1][0]
elif field == "s6":
radar_val = radar[1][7] / radar[1][0]
elif field == "s7":
radar_val = radar[1][8] / radar[1][0]
elif field == "s8":
radar_val = radar[1][9] / radar[1][0]
if radar_val > radar_threshold[1]:
return radar_val, True
else:
return radar_val, False
def ProcessQueue():
#here we are looking for alarm conditions in data
global in_queue
if in_queue:
while in_queue:
try:
tim, topic, messagein = in_queue.pop(0)
#print(".", end="")
MAC = topic[1:]
if MAC in mac_to_device_id:
device_id = mac_to_device_id[MAC]
deployment_id = device_to_deployment[device_id]
if deployment_id in alarms_settings_all:
alarms_settings = alarms_settings_all[deployment_id]
alarm_armed_settings_str = alarms_settings["enabled"]
# bit 2: Burglar Alarm (index 0), bit 1: Time Alone Alarm is Set (index 1), bit 0: Time alone Warning is Set (index 2)
if device_id in device_alerts_all:
device_alarm_settings = device_alerts_all[device_id]
message_dict = json.loads(messagein.decode('utf-8'))
if "enabled_alarms" in device_alarm_settings:
#Lets check temperatures
temp_offset = -16.0
if "1" in device_alarm_settings["enabled_alarms"][BitIndex(7):BitIndex(3)]: #Temperatures Too High/Low trigger used?
enabled_alarms_str = device_alarm_settings["enabled_alarms"]
armed_states = device_alarm_settings["armed_states"] #1=armed 0=triggered
#print(alarm_armed_settings_str)
if "temperature" in message_dict:
temperature = message_dict["temperature"] + temp_offset
#at this point temperature is in C
if temperature > 0 and temperature < 100: #ignore others
if device_id in device_alerts_all:
'''
{
"enabled_alarms":"000000000101",
"armed_states":"000000000000",
"stuck_minutes_warning":"771.3",
"stuck_warning_method_0":"SMS",
"stuck_minutes_alarm":600,
"stuck_alarm_method_1":"PHONE",
"absent_minutes_warning":"-1013.4",
"absent_warning_method_2":"SMS",
"absent_minutes_alarm":30,
"absent_alarm_method_3":"PHONE",
"temperature_high_warning":"85",
"temperature_high_warning_method_4":
"SMS","temperature_high_alarm":"95",
"temperature_high_alarm_method_5":"PHONE",
"temperature_low_warning":"60",
"temperature_low_warning_method_6":"SMS",
"temperature_low_alarm":"50",
"temperature_low_alarm_method_7":"PHONE",
"radar_alarm_method_8":"MSG",
"pressure_threshold":"15",
"pressure_alarm_method_9":"MSG",
"light_threshold":"15",
"light_alarm_method_10":"MSG",
"smell_alarm_method_11":"EMAIL",
"rearm_policy":"At midnight"
}
'''
#bits in enabled_alarms and armed_states explained
#"stuck_warning_method_0": "SMS", -1
#"stuck_alarm_method_1": "PHONE", -2
#"absent_warning_method_2": "SMS", -3
#"absent_alarm_method_3": "PHONE", -4
#"temperature_high_warning_method_4": "SMS", -5
#"temperature_high_alarm_method_5": "PHONE", -6
#"temperature_low_warning_method_6": "SMS", -7
#"temperature_low_alarm_method_7": "PHONE", -8
#"radar_alarm_method_8":"MSG", -9
#"pressure_alarm_method_9":"MSG", -10
#"light_alarm_method_10":"MSG" -11
#"smell_alarm_method_11":"MSG" -12
#hast ot be enabled and not triggerred to continue comparing
if enabled_alarms_str[BitIndex(5)] == "1" and armed_states[BitIndex(5)] == "1": #Temperatures Too High Alarm!
if temperature > FahrenheitToCelsius(device_alarm_settings["temperature_high_alarm"]):
#cancel alarm and warning, until re-armed
armed_states = set_character(armed_states, 5, "0")
armed_states = set_character(armed_states, 4, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["temperature_high_alarm_method_5"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"Alarm @ {first_last_name}: Temperature too high! ({temperature:.1f} C) in {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
#at this point we also need to store it to DB, otherwise when HTML GUI is loaded fact that trigger happened will not be reflected!
elif enabled_alarms_str[BitIndex(4)] == "1" and armed_states[BitIndex(4)] == "1": #Temperatures Too High Warning!
if temperature > FahrenheitToCelsius(device_alarm_settings["temperature_high_warning"]):
#cancel warning, until re-armed
armed_states = set_character(armed_states, 4, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["temperature_high_warning_method_4"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"Warning @ {first_last_name}: Temperature too high! ({temperature:.1f} C) in {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
if armed_states[BitIndex(7)] == "1" and armed_states[BitIndex(7)] == "1": #Temperatures Too Low Alarm!
if temperature < FahrenheitToCelsius(device_alarm_settings["temperature_low_alarm"]):
#cancel alarm and warning, until re-armed
armed_states = set_character(armed_states, 7, "0")
armed_states = set_character(armed_states, 6, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["temperature_low_alarm_method_7"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"Alarm @ {first_last_name} Temperature too low! ({temperature:.1f} C) in {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
elif armed_states[BitIndex(6)] == "1" and armed_states[BitIndex(6)] == "1": #Temperatures Too Low Warning!
if temperature < FahrenheitToCelsius(device_alarm_settings["temperature_low_warning"]):
#cancel warning, until re-armed
armed_states = set_character(armed_states, 6, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["temperature_low_warning_method_6"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"Warning @ {first_last_name} Temperature too low! ({temperature:.1f} C) in {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
#logger.info(f"{tim}, {mac}, {temperature}")
else: #radar packet
pass
#lets check if alarm condition
if alarm_armed_settings_str[BitIndex(2)] == "1": #alarm is armed!
if "radar" in message_dict:
enabled_alarms_str = device_alarm_settings["enabled_alarms"]
armed_states = device_alarm_settings["armed_states"] #1=armed 0=triggered
#if device_id == 534:
# print("Here")
if enabled_alarms_str[BitIndex(8)] == "1" and armed_states[BitIndex(8)] == "1":
radar = message_dict["radar"]
radar_threshold_str = GetRedisString(f"radar_threshold{device_id}")
radar_threshold = ast.literal_eval(radar_threshold_str)
radar_value, larger = RadarLarger(radar, radar_threshold)
print(device_id, radar_value, larger)
if larger:
#cancel alarm for this room, until re-armed
armed_states = set_character(armed_states, 8, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["radar_alarm_method_8"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"At {first_last_name} burglar alarm detected presence in the {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
if "pressure" in message_dict:
bit_nr = 9
enabled_alarms_str = device_alarm_settings["enabled_alarms"]
armed_states = device_alarm_settings["armed_states"] #1=armed 0=triggered
if enabled_alarms_str[BitIndex(bit_nr)] == "1" and armed_states[BitIndex(bit_nr)] == "1":
if message_dict["mtype"] == 1:
pressure = message_dict["pressure"]
pressure_threshold = float(device_alarm_settings["pressure_threshold"]) #GetRedisString(f"radar_threshold{device_id}")
print(device_id, pressure, pressure_threshold)
if pressure > pressure_threshold:
#cancel alarm for this room, until re-armed
armed_states = set_character(armed_states, bit_nr, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["pressure_alarm_method_9"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"At {first_last_name} burglar alarm detected doors opening in the {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
if "light" in message_dict:
bit_nr = 10
enabled_alarms_str = device_alarm_settings["enabled_alarms"]
armed_states = device_alarm_settings["armed_states"] #1=armed 0=triggered
if enabled_alarms_str[BitIndex(bit_nr)] == "1" and armed_states[BitIndex(bit_nr)] == "1":
if message_dict["mtype"] == 4:
light = message_dict["light"]
light_threshold = float(device_alarm_settings["light_threshold"]) #GetRedisString(f"radar_threshold{device_id}")
print(device_id, light, light_threshold)
if light > light_threshold:
#cancel alarm for this room, until re-armed
armed_states = set_character(armed_states, bit_nr, "0")
dev_det = devices_details_map[device_id]
location = (dev_det[4] + " " + dev_det[5].strip()).strip()
method = device_alarm_settings["light_alarm_method_10"]
first_last_name = GetBeneficiaryFromDeployment(deployment_id)
SendAlerts(deployment_id, method, f"At {first_last_name} burglar alarm detected light change in the {location}", "", "")
device_alarm_settings["armed_states"] = armed_states
device_alerts_all[device_id] = device_alarm_settings
StoreAllToRedisAndDB(deployment_id, device_id)
else:
pass #alarm not setup for this device
#print(f"{deployment_id} not in {alarms_settings_all}")
else:
pass
#logger.error(f"MAC: {mac} not part of any deployment")
except Exception as e:
logger.error(f"Error: {str(e)} {traceback.format_exc()}")
def ProcessQueueM():
#global queue
global in_queueM, PROCESSED_COUNTER, in_packets_count
#print(".")
if in_queueM:
#queue=qinfo.Open(2,0) # Open a ref to queue
#msgm=win32com.client.Dispatch("MSMQ.MSMQMessage")
while in_queueM:
#if True:
try:
topic, messagein = in_queueM.pop(0)
messagein = messagein.decode('utf-8').upper()
logger.info(f"Processing: {topic} {messagein}")
if messagein != "PIN|OK": #ignore this one
mac_pattern = r'([0-9A-Fa-f]{12})'
match = re.search(mac_pattern, messagein)
MAC = match.group(1)
record = {
'time': time.time(),
'result': messagein
}
# Convert dictionary to JSON string for storage in Redis list
record_json = json.dumps(record)
logger.info(f"Pushing to REDIS: from_alerter_{MAC} {record_json}")
redis_conn.lpush(f'from_alerter_{MAC}', record_json)
# ProcessPacket(new_message_dict)
except Exception as err:
logger.error(err)
def MAC_from_id(device_id):
conn = get_db_connection()
sql = f"SELECT device_mac FROM public.devices WHERE device_id = {device_id}"
with conn.cursor() as cur:
cur.execute(sql)
result = cur.fetchone() #cur.fetchall()
if result != None:
return result[0]
else:
return ""
def CheckRedisMessages():
global reload_needed
requests_count = 0
#print(f"CheckRedisMessages")
# Check if queue exists and has items. These items are from manual GUI interactions in alerts page
#Any test messages requested to be sent?
queue_length = redis_conn.llen('send_requests')
if queue_length > 0:
# Process each item
for i in range(queue_length):
item_json = redis_conn.rpop('send_requests')
#logger.info(f"Received in REDIS: send_requests {item_json}")
print(f"Received in REDIS: send_requests {item_json}")
if item_json is None:
break
try:
record = json.loads(item_json)
requests_count += 1
# Print the record
action = ""
logger.info(f"Request #{requests_count}:")
for key, value in record.items():
logger.info(f" {key}: {value}")
function = ""
if "function" in record:
function = record["function"]
if function == "set_group":
group_id = record["group_id"]
if "mac" in record:
MAC = record["mac"].upper()
elif "device_id" in record:
device_id = record["device_id"]
MAC = MAC_from_id(device_id).upper()
content = UNLOCK
MQSendM(f"/{MAC}", content)
time.sleep(2)
content = f"Z|{group_id}"
MQSendM(f"/{MAC}", content)
elif function == "reboot":
if "mac" in record:
MAC = record["mac"].upper()
elif "device_id" in record:
device_id = record["device_id"]
MAC = MAC_from_id(device_id).upper()
content = UNLOCK
MQSendM(f"/{MAC}", content)
time.sleep(2)
content = "s"
MQSendM(f"/{MAC}", content)
record = {
'time': time.time(),
'result': "Reboot sent..."
}
# Convert dictionary to JSON string for storage in Redis list
record_json = json.dumps(record)
logger.info(f"Pushing to REDIS: from_alerter_{MAC} {record_json}")
redis_conn.lpush(f'from_alerter_{MAC}', record_json)
elif function == "set_well_id":
if "mac" in record:
MAC = record["mac"].upper()
else:
device_id = record["device_id"]
MAC = MAC_from_id(device_id).upper()
well_id = record["well_id"]
content = UNLOCK
MQSendM(f"/{MAC}", content)
time.sleep(2)
content = f"-|{well_id}"
MQSendM(f"/{MAC}", content)
elif function == "get_device_live":
if "mac" in record:
MAC = record["mac"].upper()
else:
device_id = record["device_id"]
MAC = MAC_from_id(device_id).upper()
content = UNLOCK
MQSendM(f"/{MAC}", content)
time.sleep(2)
content = "r"
MQSendM(f"/{MAC}", content)
elif function == "set_network_id":
if "mac" in record:
MAC = record["mac"].upper()
else:
device_id = record["device_id"]
MAC = MAC_from_id(device_id).upper()
network_id = record["network_id"]
content = UNLOCK
MQSendM(f"/{MAC}", content)
time.sleep(2)
content = f"0|{network_id}"
MQSendM(f"/{MAC}", content)
else:
if "action" in record:
action = record["action"]
if action == "reload":
reload_needed = True
if action != "reload":
method = record["method"]
location_str = record["location"]
location = location_str.split("_")[2]
deployment_id = record["deployment_id"]
content = record["content"]
feature = record["feature"]
enabledCellContent = record["enabledCellContent"]
currentUnits = record["currentUnits"]
currentAlertTableMode = record["currentAlertTableMode"] #Warning/Alarm
test_only = record["test_only"]
user_name = record["user_name"]
user_first_last_name = GetBeneficiaryFromDeployment(deployment_id)
if feature == "stuck":
msg_ext = f"{currentAlertTableMode}: {content} {user_first_last_name} is spending more than {enabledCellContent} in {location}"
elif feature == "absent":
msg_ext = f"{currentAlertTableMode}: {content} {user_first_last_name} did not visit {location} in more than {enabledCellContent[1:-1]} {currentUnits}"
elif feature == "tempLow":
msg_ext = f"{currentAlertTableMode}: {content} temperature is lower then {enabledCellContent} {currentUnits} in {location} at {user_first_last_name}"
elif feature == "tempHigh":
msg_ext = f"{currentAlertTableMode}: {content} temperature is higher then {enabledCellContent} {currentUnits} in {location} at {user_first_last_name}"
elif feature == "pressure":
msg_ext = f"{currentAlertTableMode}: {content} door was opened or closed in the {location} at {user_first_last_name}"
elif feature == "radar":
msg_ext = f"{currentAlertTableMode}: {content} motion detected in the {location} at {user_first_last_name}"
else:
msg_ext = f"{currentAlertTableMode}: {content} {feature} in {location} at {user_first_last_name}"
SendAlerts(deployment_id, method, msg_ext, "Test message", user_name)
#these are testing messages, so do not count them as real triggered... so do not update in REDIS
#StoreLastSentToRedis(deployment_id)
print("-" * 40)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON from queue item: {e}")
continue
#Any test messages requested to be sent?
queue_length = redis_conn.llen('new_alarms')
if queue_length > 0:
for i in range(queue_length):
print(f"Processing send_requests message from queue...")
item_json = redis_conn.rpop('new_alarms')
logger.info(f"Received in REDIS: new_alarms {item_json}")
if item_json is None:
break
try:
record = json.loads(item_json)
deployment_id = int(record["deployment_id"])
device_id = int(record["device_id"])
if device_id == 0:
reload_needed = True
else:
print(record)
device_alarms_json = GetRedisString('alarm_device_settings_'+str(device_id))
deployment_alarms_json = GetRedisString('alarm_deployment_settings_'+str(deployment_id))
alarms_settings_all[deployment_id] = json.loads(deployment_alarms_json)
device_alerts_all[device_id] = json.loads(device_alarms_json)
print(device_alarms_json)
print(deployment_alarms_json)
#method = record["method"]
#location_str = record["location"]
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON from queue item: {e}")
continue
#print(f"Total requests processed: {requests_count}")
return requests_count
def ReloadIfNeeded():
global reload_needed, t1, t2
if reload_needed:
t1.pause()
t2.pause()
ok = load_device_configurations()
print(f"load_device_configurations in ReloadIfNeeded! {ok}")
reload_needed = False
t1.resume()
t2.resume()
# --- Main Execution ---
if __name__ == "__main__":
logger.info(f"Starting Well Alert Monitoring Service (PID: {os.getpid()})...")
logger.info(f"Alert Timeout Threshold: {TIME_OUT_THRESHOLD_MIN} minutes")
logger.info(f"Check Interval: {CHECK_INTERVAL_SECONDS} seconds")
# Register signal handlers
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
#UpdateACL("artia", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6ImFydGlhIiwiZXhwIjoxNzQ0NzY4NDYzfQ.-DFXbAeXTpKt4a-rGAPzOqcV6HBho0264qIOfZ3QdZM")
#SetupUserEnvironment("artia", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6ImFydGlhIiwiZXhwIjoxNzQ0NzcyNjQ2fQ.cu5CUWBn_6cwj5hidP61vRlr-GZwi08bgxeRFxiU2fI")
if not load_device_configurations():
logger.critical("Failed to load initial device configurations. Exiting.")
# Clean up potentially established connections
sys.exit(1)
t1 = perpetualTimer(1, ensure_mqtt_connection)
t2 = perpetualTimer(.01, ProcessQueue)
t3 = perpetualTimer(1, ReloadIfNeeded)
t4 = perpetualTimer(1, ensure_mqtt_connectionM)
t5 = perpetualTimer(.01, ProcessQueueM)
SetupTasks()
t1.start()
t2.start()
t3.start()
t4.start()
t5.start()
schedule_task(task_a, "first", 1)
schedule_task(task_b, "second", 1)
musr = os.getenv('GU')
#print(musr)
sender = EmailSender(os.getenv('GU'), os.getenv('GP'), max_workers=3)
sender.start_workers()
#SendAlerts(21, "MSG", f"Test: User way too long ({120} minutes) in Bathroom", "")
SendAlerts(21, "EMAIL", f"well-alert was started", "Program started", "robster")
#SendAlerts(21, "SMS", f"Test: User way too long ({120} minutes) in Bathroom", "", "")
#SendAlerts(21, "PHONE", f"Test: User way too long ({120} minutes) in Bathroom", "")
SendPhoneCall("-4086462191", "Hi Robert. How are you?")
#SendPhoneCall("4086462191", "Hi Robert. How are you?", "")
#SendPhoneCall("4085505424", "Hi Fred. How are you? Are you hungry?", "")
#SendPhoneCall("4087055709", "Hi Bernhard. How are you? Are you hungry?", "")
#SendPhoneCall("4086903883", "Hi Sasha. How are you? Are you hungry?", "")
#SendPhoneCall("6507968313", "Hi Julica. How are you? Are you hungry?", "") #NOT WORKING!!!
#SendPhoneCall("+385981919229", "Hi Danko. Are you hungry?", "")
#SendPhoneCall("4086462191", "Hi Danko. How are you? Your mom seems stable.", "")
#success = SendSMSTo("(408)646-2191", "This is only a test")
#time.sleep(10)
#content = UNLOCK
#MQSendM("/901506CA3C7C", content)
#time.sleep(2)
#content = "Z|1"
#MQSendM("/901506CA3C7C", content)
template_html = read_file("welcome_template_short.html")
try:
# Start the main monitoring loop
monitor_thread = threading.Thread(target=monitor_devices, daemon=True)
monitor_thread.start()
monitor_m_thread = threading.Thread(target=monitor_messages, daemon=True)
monitor_m_thread.start()
# Keep the main thread alive until stop_event is set by signal or error
while not stop_event.is_set():
# Can add periodic health checks here if needed
CheckRedisMessages()
# Optional: Check if threads are still alive and restart if needed
if not monitor_thread.is_alive():
logger.error("Device monitor thread died! Restarting...")
monitor_thread = threading.Thread(target=monitor_devices, daemon=True)
monitor_thread.start()
if not monitor_m_thread.is_alive():
logger.error("Message monitor thread died! Restarting...")
monitor_m_thread = threading.Thread(target=monitor_messages, daemon=True)
monitor_m_thread.start()
time.sleep(1) # Check stop_event periodically
logger.info("Stop event received, waiting for monitoring threads to finish...")
# Wait for both threads to finish
monitor_thread.join(timeout=CHECK_INTERVAL_SECONDS + 10)
if monitor_thread.is_alive():
logger.warning("Device monitoring thread did not finish cleanly.")
monitor_m_thread.join(timeout=10)
if monitor_m_thread.is_alive():
logger.warning("Message monitoring thread did not finish cleanly.")
except Exception as e:
logger.exception(f"An unexpected critical error occurred in main execution: {e}")
finally:
logger.info("Well Alert Monitoring Service finished.")
sys.exit(0) # Explicitly exit with success code after cleanup
print("Last line") # This will never execute due to sys.exit(0) above