How to Build a Real-Time Data Analytics Platform with Snowflake and AWS

				
					import json
import random
import time
import uuid
import boto3
import os
from datetime import datetime
from dotenv import load_dotenv
import sys


# Load environment variables from .env file
load_dotenv()


# Update the client creation to specify region


def send_to_kinesis(data, stream_name):
    # Pretty print the data being sent for debugging/monitoring purposes
    #print(json.dumps(data, indent=4))
   
    # Initialize Kinesis client with AWS credentials and region from environment variables
    # This client will handle all interactions with AWS Kinesis service
    kinesis_client = boto3.client(
        'kinesis',
        region_name=os.getenv('AWS_REGION')
    )
   
    try:
        # Attempt to send the record to Kinesis stream
        # Parameters:
        # - StreamName: The name of your Kinesis stream
        # - Data: The actual data to send (must be converted to JSON string)
        # - PartitionKey: Determines which shard the data goes to (random number used here)
        response = kinesis_client.put_record(
            StreamName=stream_name,
            Data=json.dumps(data),
            PartitionKey=str(random.randint(1, 100))  # Random partition key between 1-100
        )
       
        # Return the response from AWS which includes:
        # - SequenceNumber: Unique identifier for the record
        # - ShardId: The shard where the data was written
        return response
   
    except Exception as e:
        # If any error occurs during the sending process:
        # 1. Print the error message
        # 2. Re-raise the exception to be handled by the calling function
        print(f"Error sending record to Kinesis: {str(e)}")
        raise


def main():
   
    #Simulate  data for 10 different devices
    devices=[
        {
            "ID": 28765192,
            "IP": "149.218.116.155",
            "TYPE": "SmartHeatlhMonitor"
        }
    	#add random data for 9 other different devices
    ]
    # Iterate through each device in our devices dictionary/list
    for i in devices:
        # Create a data payload for each sensor/device
        # This represents one reading from a smart health monitoring device
        sample_data = {
            # Basic device identification
            "sensorId": i["ID"],              # Unique identifier for each sensor
            "sensorType": i["TYPE"],          # Model/type of the health monitor
            "internetIP": i["IP"],            # IP address of the device
           
            # Timestamp of the reading in ISO format (YYYY-MM-DDTHH:MM:SS)
            "connectionTime": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
           
            # Vital signs measurements with their respective units
            # Body temperature (range for simulator: 95-100°F)
            "currentTemperature": random.randint(95, 100),
            "currentTemperatureUnit": 'F',
           
            # Heart rate (range for simulator: 50-100 bpm)
            "currentPulseRate": random.randint(50, 100),
            "currentPulseRateUnit": 'Beats/Minute',
           
            # Blood oxygen level (range for simulator: 95-100%)
            "currentSpO2Level": random.randint(95, 100),
            "currentSpO2LevelUnit": '%',
           
            # Breathing rate (range for simulator: 10-15 breaths per minute)
            "currentRespiratoryRate": random.randint(10, 15),
            "currentRespiratoryRateUnit": 'Breaths/Minute',
           
            # Blood pressure readings
            # Systolic (range for simulator: 100-150 mmHg)
            "currentSystolicBloodPressure": random.randint(100, 150),
            "currentSystolicBloodPressureUnit": 'mmHg',
           
            # Diastolic (range for simulator: 60-90 mmHg)
            "currentDiastolicBloodPressure": random.randint(60, 90),
            "currentDiastolicBloodPressureUnit": 'mmHg'    
        }
   
        try:
            # Attempt to send the generated data to AWS Kinesis
            # Stream name is retrieved from environment variables
            response = send_to_kinesis(sample_data, os.getenv('KINESIS_STREAM_NAME'))
            print("Received Data for Sensor ID: ", i["ID"])
            # Response contains sequence number and shard ID (commented out to reduce console clutter)
            #print(response)
           
        except Exception as e:
            # If any error occurs during sending, print the error message
            print(f"Error sending record to Kinesis: {e}")


if __name__ == "__main__":
    # Display startup message to indicate the simulator is beginning
    print("Starting Smart Health Monitor Device Signal Simulator...")
    # Inform user how to gracefully terminate the program
    print("Press Ctrl+C to stop the program")
   
    try:
        # Infinite loop to continuously simulate device signals
        while True:
            # Execute the main function which generates and sends data for all devices
            main()
           
            # Pause execution for 30 seconds
            # This simulates real-world device behavior where vital signs
            # are measured and transmitted at regular intervals
            # Note: time.sleep() can be interrupted by Ctrl+C
            time.sleep(10)
           
    except KeyboardInterrupt:
        # Handle the Ctrl+C keyboard interrupt
        # This provides a graceful shutdown when user wants to stop the simulator
        print("\nProgram stopped by user")
        # Exit with status code 0 indicating successful/normal termination
        sys.exit(0)
       
    except Exception as e:
        # Catch any other unexpected errors that weren't specifically handled
        # This is the last line of defense for error handling
        print(f"Unexpected error: {str(e)}")
        # Exit with status code 1 indicating abnormal termination
        sys.exit(1)