Recently, I worked with a Python programmer who wanted to create a stand-alone, old-school Site Event Buffer (SEB) for a property. For those unfamiliar, a SEB is a device designed to monitor, capture, and report events and alarms, often from telecommunication or building management systems.
While many systems now have built-in notification features, an SEB is a different tool. Rather than just alerting issues, it continuously collects, analyzes, and stores data, much like a “black box” in aviation.
In the event of a system failure, the SEB provides invaluable insights by logging the events leading up to the incident, helping administrators piece together what went wrong. Additionally, SEB data can be accessed and analyzed remotely, allowing administrators to anticipate what equipment might be needed on-site.
For example, if a device starts showing power supply errors before shutting down completely, the SEB will capture these critical events. Armed with this information, the administrator could determine that a replacement power supply is likely required before arriving at the property, making troubleshooting more efficient and minimizing downtime.
This SEB project took an interesting turn partway through development. Although the programmer was initially building the SEB for one property, he soon realized he wanted to replicate it across four additional properties. The catch? While the systems he planned to monitor were all the same brand, they varied by version, with each version sending slightly different data. This variance presented a challenge: the programmer’s initial design required a fixed schema, which wouldn’t adapt easily to different system versions.
To address this, I introduced the concept of “letting the software define the structure of the database.” This approach allows the SEB to adapt dynamically, modifying its database structure based on the data it encounters, rather than relying on a fixed schema.
SIDE NOTE: The concept of “letting the software define the structure of the database” aligns closely with “schema-on-read” or “dynamic schema” approaches in database and data management terminology.
Here’s how each term applies:
- Schema-on-Read: This approach, often used in NoSQL databases and data lakes, allows the structure or schema to be defined at the time of reading the data rather than at the time of writing it. The software adapts to the data structure dynamically as it retrieves data, enabling flexibility in handling different data formats or versions. In the context of a SEB, schema-on-read would allow it to adjust its structure each time it interacts with a new version of the system data.
- Dynamic Schema: This term refers to a database schema that can change dynamically as new fields or types of data are introduced. This approach is common in systems that interact with varying data sources, as it allows the database to evolve without predefined structures. In SQL databases, dynamic schemas often require modifications to adapt to new data structures, while NoSQL databases like MongoDB natively support this approach.
Let’s walk through how this flexible approach can be implemented and why it’s ideal for projects like a Site Event Buffer. We’ll work through the fictitious example below — a Uninterruptible Power Supply (UPS).
Step 1: Initial Query — Retrieving Fault Data Structure via USB
To start, the SEB needs to determine which fields each UPS version will provide. Rather than predefining a schema, the SEB connects to the UPS over USB and sends a command — let’s call it "Show Faults"—to retrieve the fields directly. The SEB will then inspect the returned data to build its database structure dynamically.
Below, we simulate the process with Python code. Here’s how the SEB would send the command and interpret the response:
import sqlite3
# Function to simulate sending a command to UPS via USB and getting fault fields
def send_usb_command(command):
if command == "Show Faults":
# Simulating response from UPS Version 1.0
return {
"timestamp": "2024-11-14T12:34:56",
"fault_code": 101,
"battery_status": "Faulty",
"temperature": 45.0,
"load_percentage": 80
}
# Send the command to retrieve initial fault data
initial_fault_data = send_usb_command("Show Faults")In this example, the SEB receives five fields from the UPS: timestamp, fault_code, battery_status, temperature, and load_percentage. Based on this response, the SEB will create a database table with corresponding fields.
Step 2: Building the Initial Database Schema
Now, with the fault data fields retrieved, the SEB can create a database table to store this information. Using SQLite, we dynamically create a table called ups_faults that matches the fields provided by the UPS. Here’s the code to establish this initial schema:
# Connect to the database (or create it if it doesn't exist)
conn = sqlite3.connect("ups_fault_data.db")
cursor = conn.cursor()
# Function to dynamically create or update the table schema based on the fault data keys
def ensure_table_schema(data):
# Create a base table if it doesn't already exist (with an ID column)
cursor.execute("""
CREATE TABLE IF NOT EXISTS ups_faults (
id INTEGER PRIMARY KEY AUTOINCREMENT
)
""")
# Dynamically add columns for each key in the data dictionary
for key, value in data.items():
# Determine the SQLite column type based on the Python type
if isinstance(value, int):
column_type = "INTEGER"
elif isinstance(value, float):
column_type = "REAL"
elif isinstance(value, str):
column_type = "TEXT"
else:
column_type = "TEXT" # Default to TEXT for unsupported types
# Add the column if it doesn't already exist
try:
cursor.execute(f"ALTER TABLE ups_faults ADD COLUMN {key} {column_type}")
except sqlite3.OperationalError:
# Column already exists
pass
# Ensure the schema matches the structure of the initial fault data
ensure_table_schema(initial_fault_data)
# Dynamically insert the initial fault data into the table
def insert_data(data):
# Build dynamic SQL query for insertion
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data.values())
query = f"INSERT INTO ups_faults ({keys}) VALUES ({placeholders})"
# Execute the query with the data values
cursor.execute(query, list(data.values()))
# Insert the initial data
insert_data(initial_fault_data)
# Commit changes and close the connection
conn.commit()
conn.close()
print("Database created/updated and initial data inserted successfully!")The Full Code:
import sqlite3
# Function to simulate sending a command to UPS via USB and getting fault fields
def send_usb_command(command):
if command == "Show Faults":
# Simulating response from UPS Version 1.0
return {
"timestamp": "2024-11-14T12:34:56",
"fault_code": 101,
"battery_status": "Faulty",
"temperature": 45.0,
"load_percentage": 80
}
# Send the command to retrieve initial fault data
initial_fault_data = send_usb_command("Show Faults")
# Connect to the database (or create it if it doesn't exist)
conn = sqlite3.connect("ups_fault_data.db")
cursor = conn.cursor()
# Function to dynamically create or update the table schema based on the fault data keys
def ensure_table_schema(data):
# Create a base table if it doesn't already exist (with an ID column)
cursor.execute("""
CREATE TABLE IF NOT EXISTS ups_faults (
id INTEGER PRIMARY KEY AUTOINCREMENT
)
""")
# Dynamically add columns for each key in the data dictionary
for key, value in data.items():
# Determine the SQLite column type based on the Python type
if isinstance(value, int):
column_type = "INTEGER"
elif isinstance(value, float):
column_type = "REAL"
elif isinstance(value, str):
column_type = "TEXT"
else:
column_type = "TEXT" # Default to TEXT for unsupported types
# Add the column if it doesn't already exist
try:
cursor.execute(f"ALTER TABLE ups_faults ADD COLUMN {key} {column_type}")
except sqlite3.OperationalError:
# Column already exists
pass
# Ensure the schema matches the structure of the initial fault data
ensure_table_schema(initial_fault_data)
# Dynamically insert the initial fault data into the table
def insert_data(data):
# Build dynamic SQL query for insertion
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data.values())
query = f"INSERT INTO ups_faults ({keys}) VALUES ({placeholders})"
# Execute the query with the data values
cursor.execute(query, list(data.values()))
# Insert the initial data
insert_data(initial_fault_data)
# Commit changes and close the connection
conn.commit()
conn.close()
print("Database created/updated and initial data inserted successfully!")Code creates a database table with fields that perfectly match the data provided by the first UPS model (Version 1.0). But what happens if a different version with additional fields connects?
Step 3: Adapting to a New UPS Version
Later, the SEB encounters another UPS, this time running a newer firmware version (let’s call it Version 2.5). When the SEB sends the same "Show Faults" command, the response now includes additional fields: battery_voltage and input_voltage. Here’s how this response might look:
# Simulated function for UPS Version 2.5 response
def send_usb_command_v2(command):
if command == "Show Faults":
return {
"timestamp": "2024-11-14T12:35:20",
"fault_code": 201,
"battery_status": "Faulty",
"temperature": 47.0,
"load_percentage": 85,
"battery_voltage": 12.5,
"input_voltage": 220
}
# Retrieve fault data fields from UPS Version 2.5
new_fault_data = send_usb_command_v2("Show Faults")With new fields now in the returned data, the SEB will adapt its database to store the additional battery_voltage and input_voltage fields. Here’s how it dynamically alters the schema to accommodate these new fields:
import sqlite3
# Simulated function for UPS Version 2.5 response
def send_usb_command_v2(command):
if command == "Show Faults":
return {
"timestamp": "2024-11-14T12:35:20",
"fault_code": 201,
"battery_status": "Faulty",
"temperature": 47.0,
"load_percentage": 85,
"battery_voltage": 12.5,
"input_voltage": 220
}
# Function to ensure the table schema matches the data dynamically
def ensure_table_schema(conn, table_name, data):
cursor = conn.cursor()
# Create the table if it doesn't exist with a base column
cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {table_name} (
id INTEGER PRIMARY KEY AUTOINCREMENT
)
""")
# Add columns dynamically if they don't exist
for key, value in data.items():
# Determine column type based on data type
if isinstance(value, int):
column_type = "INTEGER"
elif isinstance(value, float):
column_type = "REAL"
elif isinstance(value, str):
column_type = "TEXT"
else:
column_type = "TEXT" # Default to TEXT for unknown types
# Attempt to add the column; skip if it already exists
try:
cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {key} {column_type}")
except sqlite3.OperationalError:
# Column already exists
pass
conn.commit()
# Function to insert data dynamically into the table
def insert_data(conn, table_name, data):
cursor = conn.cursor()
# Build dynamic SQL query for insertion
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data.values())
query = f"INSERT INTO {table_name} ({keys}) VALUES ({placeholders})"
# Execute the query with the data values
cursor.execute(query, list(data.values()))
conn.commit()
# Main function to process fault data dynamically
def process_fault_data(data, table_name="ups_faults"):
# Connect to the database
conn = sqlite3.connect("ups_fault_data.db")
# Ensure the table schema matches the current data
ensure_table_schema(conn, table_name, data)
# Insert the data into the table
insert_data(conn, table_name, data)
# Close the connection
conn.close()
# Example usage
# Process data from UPS Version 1.0
initial_fault_data = send_usb_command_v1("Show Faults")
process_fault_data(initial_fault_data)
# Process data from UPS Version 2.5
new_fault_data = send_usb_command_v2("Show Faults")
process_fault_data(new_fault_data)
print("Database updated successfully!")Explanation and Benefits of the Code
- Flexible Schema Initialization: The SEB first creates a table based on fields from the initial UPS model it connects to, establishing a foundational schema.
- Adaptive Schema Updates: When a new UPS model with additional fields is encountered, the SEB adjusts its schema by adding the new fields only if they don’t already exist. This ensures compatibility with both old and new UPS versions without requiring separate tables or manual intervention.
- Backward and Forward Compatibility: This approach allows the SEB to store fault data from multiple versions of UPS systems, making the solution both backward and forward-compatible.
Code to upload data to the database every five minutes. We’ll use a loop with a five-minute delay to simulate continuous monitoring, where the SEB polls the UPS for fault data using the "Show Faults" command and uploads any new data to the database.
import sqlite3
import time
from datetime import datetime
# Simulated function for retrieving fault data from the UPS
def get_fault_data():
# Simulating UPS response with random data for example purposes
return {
"timestamp": datetime.now().isoformat(),
"fault_code": 101, # Example fault code
"battery_status": "Faulty" if datetime.now().minute % 2 == 0 else "OK",
"temperature": 45.0 + datetime.now().minute % 5, # Simulating varying temperature
"load_percentage": 80 + datetime.now().second % 10, # Simulating varying load
"battery_voltage": 12.5 if datetime.now().second % 2 == 0 else 12.8,
"input_voltage": 220 + datetime.now().second % 5
}
# Function to insert fault data into the database
def insert_fault_data(conn, table_name, data):
cursor = conn.cursor()
# Build the dynamic SQL query for insertion
keys = ", ".join(data.keys())
placeholders = ", ".join("?" for _ in data.values())
query = f"INSERT INTO {table_name} ({keys}) VALUES ({placeholders})"
# Execute the query with the data values
cursor.execute(query, list(data.values()))
conn.commit()
# Continuous monitoring function to upload data every 5 minutes
def continuous_monitoring(table_name="ups_faults", interval=300):
conn = sqlite3.connect("ups_fault_data.db")
while True:
try:
# Retrieve new fault data from UPS
fault_data = get_fault_data()
# Insert the fault data into the database
insert_fault_data(conn, table_name, fault_data)
print(f"Data uploaded to database: {fault_data}")
except sqlite3.Error as e:
print(f"Database error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
# Wait for the specified interval before repeating
time.sleep(interval)
# Close the database connection (rarely reached since this loops forever)
conn.close()
# Start monitoring
continuous_monitoring()What a SEB Might Do:
Here’s a Python code example for the SEB that performs historical analysis on UPS fault data, allowing a user to query fault codes, view statistics on data collection duration, total data points, and frequency of specific fault codes. Additionally, the SEB will notify the user via SIP over the network, and if that fails, it will attempt to dial out using a USB 56K modem.
This example assumes we have a SQLite database (ups_fault_data.db) that contains a table with fault data, including fault codes and timestamps. The code includes querying, statistical analysis, and notification via network and fallback to a modem.
Setup Code: Database Connection and Helper Functions
First, we’ll set up some helper functions to connect to the database, retrieve fault codes, analyze historical data, and send notifications.
import sqlite3
from datetime import datetime
import random
import time
# Sample setup for network and modem notification
class Notifier:
def send_sip_notification(self, message):
print("Attempting to send SIP notification...")
# Simulating SIP notification success/failure
return random.choice([True, False]) # Randomly succeed or fail for example
def dial_modem(self, phone_number, message):
print(f"Dialing out via modem to {phone_number}...")
# Simulating a modem dialing action
print(f"Message sent via modem: {message}")
notifier = Notifier()
# Connect to SQLite database
conn = sqlite3.connect("ups_fault_data.db")
cursor = conn.cursor()
# Helper function to query fault code details
def query_fault_code(fault_code):
cursor.execute("SELECT * FROM ups_faults WHERE fault_code = ?", (fault_code,))
results = cursor.fetchall()
if results:
for row in results:
print(f"Fault Code: {row[1]}, Timestamp: {row[0]}, Status: {row[2]}, Temperature: {row[3]}, Load: {row[4]}")
else:
print("No data found for the specified fault code.")Historical Analysis Functions
These functions will analyze how long data has been collected, count data points, and identify common faults.
# Calculate how long data has been collected
def data_collection_duration():
cursor.execute("SELECT MIN(timestamp), MAX(timestamp) FROM ups_faults")
min_time, max_time = cursor.fetchone()
if min_time and max_time:
min_date = datetime.fromisoformat(min_time)
max_date = datetime.fromisoformat(max_time)
duration = max_date - min_date
print(f"Data collection has been ongoing for: {duration.days} days")
else:
print("No data available.")
# Count the total number of data points
def count_data_points():
cursor.execute("SELECT COUNT(*) FROM ups_faults")
total_points = cursor.fetchone()[0]
print(f"Total data points collected: {total_points}")
# Find the most frequent fault code
def most_frequent_fault_code():
cursor.execute("SELECT fault_code, COUNT(fault_code) as freq FROM ups_faults GROUP BY fault_code ORDER BY freq DESC LIMIT 1")
result = cursor.fetchone()
if result:
print(f"The most frequent fault code is {result[0]}, occurring {result[1]} times.")
else:
print("No fault data available.")Notification Logic
The SEB attempts to notify the user first via SIP. If the network is down, it will fall back to a USB 56K modem dial-out.
# Notification function with SIP and modem fallback
def notify_user(message, phone_number):
if notifier.send_sip_notification(message):
print("Notification sent successfully via SIP.")
else:
print("SIP notification failed. Attempting to dial out via modem.")
notifier.dial_modem(phone_number, message)
# Example usage: notify user with a message
def main():
# Query a specific fault code
fault_code = 101 # Example fault code
print("Querying fault code details:")
query_fault_code(fault_code)
print("\nHistorical Analysis:")
data_collection_duration()
count_data_points()
most_frequent_fault_code()
# Send notification with fallback to modem
message = "Critical fault detected in UPS system. Immediate attention required."
phone_number = "555-1234" # Example phone number
notify_user(message, phone_number)
main()Explanation of Each Part
Database Query and Fault Code Analysis:
- The
query_fault_code()function allows users to search for specific fault codes and view detailed information. data_collection_duration()calculates how long the SEB has been collecting data by finding the earliest and latest timestamps in the database.count_data_points()counts the total number of records in the database, providing an overview of data density.most_frequent_fault_code()identifies the most commonly occurring fault code, offering insight into recurring issues.
Notification with Fallback:
- The
notify_user()function attempts to send a notification via SIP first. If the SIP notification fails (simulated by random choice here), it dials out using a modem. - This provides a redundant notification system, ensuring critical alerts reach the user even if network connectivity is down.
Example Output
Querying fault code details:
Fault Code: 101, Timestamp: 2024-11-14T12:34:56, Status: Faulty, Temperature: 45.0, Load: 80
Historical Analysis:
Data collection has been ongoing for: 30 days
Total data points collected: 350
The most frequent fault code is 101, occurring 150 times.
Attempting to send SIP notification...
SIP notification failed. Attempting to dial out via modem.
Dialing out via modem to 555-1234...
Message sent via modem: Critical fault detected in UPS system. Immediate attention required.Now, the SEB is set up to independently audit the UPS system with capabilities similar to an aviation black box. Every five minutes, it collects critical data, logging the health, status, and any fault codes in a detailed timeline.
If interested in the old SEB’s, visit the link below to see an old Teltronic’s SEB II sales PDF.
https://drive.google.com/file/d/1a-dGTEP3rPHVOeQR4tlZiAh6FQ5CFVhV/view?usp=drive_link
Thank you for reading this article. I hope you found it helpful and informative. If you have any questions, or if you would like to suggest new Python code examples or topics for future tutorials, please feel free to reach out. Your feedback and suggestions are always welcome!
Happy coding!
C. C. Python Programming
You can also find this tutorial at Medium.com
