IOT Project and Flask

Raspberrypi/RGBsmoothcolor.py

+form ctypes.wintypes import RGB
+from time import sleep
+import RPi.GPIO as GPIO
+from colour import Color
+
+class RGBA():
+    def __init__(self, r, g, b):
+        GPIO.setmode(GPIO.BCM)
+        channel = [r, g, b]
+        for c in channel:
+            GPIO.setup(c, GPIO.OUT)
+        self.r = GPIO.PWM(r, 120) # channel = 12 frequency = 60Hz
+        self.g = GPIO.PWM(g, 120) # channel = 12 frequency = 60Hz
+        self.b = GPIO.PWM(b, 120) # channel = 12 frequency = 60Hz
+        self.r.start(0)
+        self.g.start(0)
+        self.b.start(0)
+        
+# 0 - 225 for (r, g, b)
+
+    def setRGBColor(self, r, g, b):
+        r = 100 - (r / 225) * 100
+        g = 100 - (r / 225) * 100
+        b = 100 - (r / 225) * 100
+        print( r, g, b )
+        
+        self.r.ChangeDutyCycle(int(r))
+        self.g.ChangeDutyCycle(int(g))
+        self.b.ChangeDutyCycle(int(b))
+        
+# 0 - 100 for colours module
+    def setRGB(self, rgb):
+        self.r.ChangeDutyCycle(int(rgb[0]))
+        self.g.ChangeDutyCycle(int(rgb[1]))
+        self.b.ChangeDutyCycle(int(rgb[2]))
+        print( r, g, b )
+        
+# 0 - 100 for colours module ( any error to use this code...)
+''' def setRGB(self, rgb):
+        r = abs(rgb[0] * 100 - 100 )
+        g = abs(rgb[1] * 100 - 100 )
+        b = abs(rgb[2] * 100 - 100 )
+        print(r, g, b)
+        self.r.ChangeDutyCycle(int(r))
+        self.g.ChangeDutyCycle(int(g))
+        self.b.ChangeDutyCycle(int(b))
+'''     
+# led = RGBA(12,13,19)
+# led.setRGBColor(255,0,0) # type any color code to blink
+# sleep(10)
+# GPIO.cleanup()
+
+#led = RGBA(12,13,19)
+# led.setRGB(Color(input("Enter Any Color...")).rgb) # Enter any color code to blink
+# sleep(10)
+# GPIO.cleanup()
+led = RGBA(12, 13, 19)
+while True:
+    for c in Color("rgb").range_to(Color("blue") ,100):
+        led.setRGB(c.rgb)
+        sleep(0.1)
+    for c in Color("blue").range_to(Color("red") ,100):
+        led.setRGB(c.rgb)
+        sleep(0.1)
\ No newline at end of file

Raspberrypi/type0-7color.py

+import RPi.GPIO as GPIO
+from time import sleep
+
+#GPIO.cleanup()
+GPIO.setmode(GPIO.BCM)
+channel = [12, 13, 19]
+for c in channel:
+    GPIO.setup(c, GPIO.OUT)
+#GPIO.output(channel, GPIO.HIGH)
+#GPIO.output(channel, GPIO.LOW)
+
+try:
+    while True:
+        try:
+            i = input("Enter a number between 0-7...")
+            i = int(i)
+            if i < 8 or i >=8:
+                print("Invalid range of input... Try again...")
+                continue
+            rgb = format(i, '03b')
+            for i, c in enumerate(channel):
+                print(i, c, rgb[i])
+                GPIO.output(c, not bool(int(rgb[i])))
+                
+        except ValueError:
+            print("Invalid Input, Try again...")
+            
+except KeyboardInterrupt:
+    GPIO.cleanup()
+    print("Quitting...")
\ No newline at end of file

create venv file/create venv file.txt

+Create a Venv File:
+    python -m venv iotcloud
+
+Activate This Venv File:
+    source iotcloud/bin/activate
+
+Linex System PC Full Information Command:
+    /proc/cpuinfo
\ No newline at end of file

flask/app.py

+from flask import Flask
+app = Flask(__name__)
+import os
+Basename = '/iotcloud'
+# app.debug = True ( for any debug error to True that line ) 
+
+
+# @app.route na searching route ex:[172.30.0.231:7000/hello] to view your flask project.
+@app.route('/hello')
+def hello_world():
+   return "<h1>Hello World 123<h1/>"
+
+@app.route('/')
+def hello():
+   return "<h1>Hello Justin Welcome<h1/>"
+
+@app.route('/whoami')
+def whoami():
+   return "<h1>"+os.popen('whoami').read()+"<h1/>"
+   # Which one you can return for this command [this is this system user name]
+
+def cpuinfo():
+   return "<pre>" "<h3>"+ os.popen('cat /proc/cpuinfo').read() +"<pre/>" "<h3/>"
+   # return = Which one you can return for this command [this is system full information]
+   # <pre> = To HTML tag Which Terminal output You Can Inset to Same output and display
+app.add_url_rule (Basename+'/whoami','whoami',whoami)
+app.add_url_rule (Basename+'/cpuinfo','cpuinfo',cpuinfo)
+
+if __name__ == '__main__':
+   app.run(host = '0.0.0.0' , port = '7000' , debug = True )
+   # your access all type of IP address to add [host = '0.0.0.0'] Running on all IP addresses
\ No newline at end of file

mongodb/test.py

+from pymongo import MongoClient
+
+client = MongoClient('mongodb://Justin_Mass:Amutha%401981@mongodb.selfmade.ninja:27017/?authSource=users')
+
+db = client.Justin_Mass_iotcloud
+# db = client['Justin_Mass_iotcloud'] alternate way
+
+result = db.test.find_one({
+  "username": "Justin"  
+})
+
+print(result)
\ No newline at end of file

python-marathon/main.py

+def add (x,y):
+    return x+y
+print(add(1,3))
\ No newline at end of file

python-marathon/snalib.py

+import main

rabbitmq/simple_consumer.py

+import pika
+def callback(ch, method, properties, body):
+    print(" [ X ] Received % r " % body )
+
+credentials = pika.PlainCredentials('Justin', 'Justin@Saran2001')
+parameters = pika.ConnectionParameters('rabbitmq.selfmade.ninja',
+                                       5672,
+                                       'justinmass2001_Amutha',
+                                       credentials)
+
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+channel.queue_declare(queue='my_first_queue')
+channel.basic_consume(queue='my_first_queue',
+                      auto_ack=True,
+                      on_message_callback=callback)
+print(' [*] Waiting for messages. To exit press CTRL+C')
+channel.start_consuming()
\ No newline at end of file

rabbitmq/simple_producer.py

+import pika
+import random
+from time import sleep
+
+credentials = pika.PlainCredentials('Justin', 'Justin@Saran2001')
+parameters = pika.ConnectionParameters('rabbitmq.selfmade.ninja',
+                                       5672,
+                                       'justinmass2001_Amutha',
+                                       credentials)
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+channel.queue_declare(queue='my_first_queue')
+
+while True:
+    message = str(random.random())
+    channel.basic_publish(exchange='', 
+                        routing_key='my_first_queue',
+                        body='Hello World!')
+    print(" [x] Sent '{}'".format(message))
+    sleep(0.01)
+#    connection.close()
+    
\ No newline at end of file