From 43dc93ff5cf88c151ea7aa41aba9f8549115fffa Mon Sep 17 00:00:00 2001 From: liam mall <29150314+liamcmall@users.noreply.github.com> Date: Mon, 1 Apr 2024 20:18:30 +0000 Subject: [PATCH 01/11] Still need to fix MongoDB connection erghhh --- app/data.py | 146 +++++++++++++++++++++++++++++++++++++++++++++++++--- app/main.py | 9 +++- 2 files changed, 147 insertions(+), 8 deletions(-) diff --git a/app/data.py b/app/data.py index 61e69e5..2b5d5cd 100644 --- a/app/data.py +++ b/app/data.py @@ -1,25 +1,159 @@ from os import getenv - +import random +from datetime import datetime +import math from certifi import where from dotenv import load_dotenv from MonsterLab import Monster from pandas import DataFrame from pymongo import MongoClient +from typing import Dict, Iterable, Iterator + +ranks = ['Private', 'Sergeant', 'Lieutenant', 'Captain', 'Commander'] +rank_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] + +clone_types = ['Standard','Recon', 'Heavy', 'Commando', 'ARC Trooper'] +clone_type_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] + +asssigned_weapons = ["DC-15A Blaster Rifle", + "DC-15X Sniper Rifle", + "DC-17m Blaster Rifle", + "DC-15S Blaster Carbine", + "WESTAR-M5 Blaster Rifle"] + +assigned_general = ['General Mace Windu', "General Plo Koon", 'General Yoda', 'General Kenobi', 'General Skywalker',] +assigned_general_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] class Database: + def __init__(self, collection: str): + self.db = MongoDB(collection) def seed(self, amount): - pass + data = self.generate_clone_trooper(amount) + self.db.create_many(data) def reset(self): - pass + self.db.delete_all() def count(self) -> int: - pass + return self.db.count_documents() def dataframe(self) -> DataFrame: - pass + return self.db.to_dataframe() def html_table(self) -> str: - pass + return self.db.to_html_table() + + @staticmethod + def generate_clone_trooper(n): + data = [] + for _ in range(n): + clone_trooper_name = f"CT-{random.randint(0, 999999):06}" + # ... rest of your code here ... + data.append({ + "CT ID": clone_trooper_name, + 'Clone Type': clone_type, + 'Rank': rank, + 'Assigned Weapon': assigned_weapon, + 'Health': health, + 'Energy': energy, + 'Success Percentage': success_percentage, + 'Assigned General': general, + 'Check In Time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') + }) + return data + + +class MongoDB: + load_dotenv() + database = MongoClient(getenv("DB_URL") + "&tls=true&tlsInsecure=true")["Database"] + # load_dotenv() + # database = MongoClient(getenv("DB_URL"), tlsCAFile=where())["Database"] + + def __init__(self, collection: str): + self.collection = self.database[collection] + + def create_one(self, record: Dict) -> bool: + return self.collection.insert_one(record).acknowledged + + def read_one(self, query: Dict) -> Dict: + return self.collection.find_one(query, {"_id": False}) + + def update_one(self, query: Dict, update: Dict) -> bool: + return self.collection.update_one(query, {"$set": update}).acknowledged + + def delete_one(self, query: Dict) -> bool: + return self.collection.delete_one(query).acknowledged + + def create_many(self, records: Iterable[Dict]) -> bool: + return self.collection.insert_many(records).acknowledged + + def read_many(self, query: Dict) -> Iterator[Dict]: + return self.collection.find(query, {"_id": False}) + + def update_many(self, query: Dict, update: Dict) -> bool: + return self.collection.update_many(query, {"$set": update}).acknowledged + + def delete_many(self, query: Dict) -> bool: + return self.collection.delete_many(query).acknowledged + + def reset(self): + self.db.delete_all() + + def count(self) -> int: + return self.db.count_documents() + + def dataframe(self) -> DataFrame: + return self.db.to_dataframe() + + def html_table(self) -> str: + return self.db.to_html_table() + + def count_documents(self) -> int: + return self.collection.count_documents({}) + +# if __name__ == '__main__': +# db = MongoDB("Collection") +# db.create_many({"Value": randrange(1, 100)} for _ in range(10)) +# print(DataFrame(db.read_many({}))) + +# import unittest +# from data import Database + +# class TestDatabase(unittest.TestCase): +# def setUp(self): +# self.db = Database('test_collection') + +# def test_seed(self): +# self.db.seed(10) +# self.assertEqual(self.db.count(), 10) + +# def test_reset(self): +# self.db.reset() +# self.assertEqual(self.db.count(), 0) + +# def test_dataframe(self): +# self.db.seed(10) +# df = self.db.dataframe() +# self.assertEqual(len(df), 10) + +# def test_html_table(self): +# self.db.seed(10) +# html = self.db.html_table() +# self.assertIsNotNone(html) + +# if __name__ == '__main__': +# unittest.main() \ No newline at end of file diff --git a/app/main.py b/app/main.py index 1f9e0b0..2996dd7 100644 --- a/app/main.py +++ b/app/main.py @@ -5,12 +5,14 @@ from MonsterLab import Monster from flask import Flask, render_template, request from pandas import DataFrame +from datetime import datetime from app.data import Database +from app.data import MongoDB from app.graph import chart from app.machine import Machine -SPRINT = 0 +SPRINT = 1 APP = Flask(__name__) @@ -28,7 +30,10 @@ def home(): def data(): if SPRINT < 1: return render_template("data.html") - db = Database() + # for personal reasons + collection_name = "Collection_" + datetime.now().strftime("%Y%m%d%H%M%S") + db = Database(collection_name) + # db = MongoDB(collection_name) return render_template( "data.html", count=db.count(), From 9982bc8fdf5921823bd01be3a97149cc0a43fc4e Mon Sep 17 00:00:00 2001 From: liam mall <29150314+liamcmall@users.noreply.github.com> Date: Thu, 4 Apr 2024 20:50:00 +0000 Subject: [PATCH 02/11] Moved from MongoDB to ElephantSQL. Test --- .vscode/settings.json | 3 + app/data.py | 171 +++++++++++++++++++--------------------- app/data2.py | 163 ++++++++++++++++++++++++++++++++++++++ app/graph.py | 1 + app/main.py | 9 +-- app/templates/data.html | 4 +- 6 files changed, 254 insertions(+), 97 deletions(-) create mode 100644 .vscode/settings.json create mode 100644 app/data2.py diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..457f44d --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "python.analysis.typeCheckingMode": "basic" +} \ No newline at end of file diff --git a/app/data.py b/app/data.py index 2b5d5cd..f4d907b 100644 --- a/app/data.py +++ b/app/data.py @@ -1,14 +1,23 @@ -from os import getenv -import random +# Standard library imports from datetime import datetime +from os import getenv import math +import random +import unittest + +# Misc imports from certifi import where from dotenv import load_dotenv -from MonsterLab import Monster -from pandas import DataFrame from pymongo import MongoClient +from pymongo.errors import ServerSelectionTimeoutError +import pandas as pd + +# Typing from typing import Dict, Iterable, Iterator + +# List and Etc. + ranks = ['Private', 'Sergeant', 'Lieutenant', 'Captain', 'Commander'] rank_probabilities = [round(random.uniform(0.0,0.1),2), round(random.uniform(0.1,0.15),2), @@ -23,7 +32,7 @@ round(random.uniform(0.25,0.3),2), round(random.uniform(0.3,0.35),2)] -asssigned_weapons = ["DC-15A Blaster Rifle", +assigned_weapons = ["DC-15A Blaster Rifle", "DC-15X Sniper Rifle", "DC-17m Blaster Rifle", "DC-15S Blaster Carbine", @@ -38,31 +47,65 @@ class Database: - def __init__(self, collection: str): - self.db = MongoDB(collection) + load_dotenv() + + def __init__(self, collection_name: str): + db_url = getenv("DB_URL") + self.client = MongoClient(db_url, + tlsCAFile=where(), + tls=True, + tlsAllowInvalidCertificates=True)["MainDatabase"] + self.collection = self.client[collection_name] def seed(self, amount): data = self.generate_clone_trooper(amount) - self.db.create_many(data) - - def reset(self): - self.db.delete_all() + self.collection.insert_many(data) + + def reset(self) -> int: + return self.collection.delete_many({}).deleted_count def count(self) -> int: - return self.db.count_documents() + return self.collection.count_documents({}) - def dataframe(self) -> DataFrame: - return self.db.to_dataframe() + def to_dataframe(self) -> pd.DataFrame: + return pd.DataFrame(list(self.collection.find())) - def html_table(self) -> str: - return self.db.to_html_table() + def to_html_table(self) -> str: + df = self.to_dataframe() + return df.to_html() + # Generate a random clone trooper! @staticmethod def generate_clone_trooper(n): data = [] for _ in range(n): clone_trooper_name = f"CT-{random.randint(0, 999999):06}" - # ... rest of your code here ... + + # Select a clone type and its probability + clone_type_index = random.choices(range(len(clone_types)), clone_type_probabilities, k=1)[0] + clone_type = clone_types[clone_type_index] + clone_type_probability = clone_type_probabilities[clone_type_index] + + # Select a rank and its probability + rank_index = random.choices(range(len(ranks)), rank_probabilities, k=1)[0] + rank = ranks[rank_index] + rank_probability = rank_probabilities[rank_index] + + # Weapon Selected + assigned_weapon = random.choice(assigned_weapons) + + # Random health and energy + health = random.randint(3, 9) + energy = random.randint(2, 6) + + # Select a general and its probability + general_index = random.choices(range(len(assigned_general)), assigned_general_probabilities, k=1)[0] + general = assigned_general[general_index] + general_probability = assigned_general_probabilities[general_index] + + # Calculated the success percentage. But also included a skew towards healthier clones + success_percentage = round(sum([clone_type_probability, rank_probability, general_probability,math.log10(health)-0.80]), 2) + data.append({ "CT ID": clone_trooper_name, 'Clone Type': clone_type, @@ -77,83 +120,33 @@ def generate_clone_trooper(n): return data -class MongoDB: - load_dotenv() - database = MongoClient(getenv("DB_URL") + "&tls=true&tlsInsecure=true")["Database"] - # load_dotenv() - # database = MongoClient(getenv("DB_URL"), tlsCAFile=where())["Database"] - - def __init__(self, collection: str): - self.collection = self.database[collection] - - def create_one(self, record: Dict) -> bool: - return self.collection.insert_one(record).acknowledged - - def read_one(self, query: Dict) -> Dict: - return self.collection.find_one(query, {"_id": False}) - - def update_one(self, query: Dict, update: Dict) -> bool: - return self.collection.update_one(query, {"$set": update}).acknowledged - - def delete_one(self, query: Dict) -> bool: - return self.collection.delete_one(query).acknowledged - - def create_many(self, records: Iterable[Dict]) -> bool: - return self.collection.insert_many(records).acknowledged - - def read_many(self, query: Dict) -> Iterator[Dict]: - return self.collection.find(query, {"_id": False}) - - def update_many(self, query: Dict, update: Dict) -> bool: - return self.collection.update_many(query, {"$set": update}).acknowledged - - def delete_many(self, query: Dict) -> bool: - return self.collection.delete_many(query).acknowledged - - def reset(self): - self.db.delete_all() - - def count(self) -> int: - return self.db.count_documents() - - def dataframe(self) -> DataFrame: - return self.db.to_dataframe() - - def html_table(self) -> str: - return self.db.to_html_table() - - def count_documents(self) -> int: - return self.collection.count_documents({}) +## TESTING CENTER ### -# if __name__ == '__main__': -# db = MongoDB("Collection") -# db.create_many({"Value": randrange(1, 100)} for _ in range(10)) -# print(DataFrame(db.read_many({}))) +class TestMongoDBConnection(unittest.TestCase): -# import unittest -# from data import Database + def setUp(self): + db_url = getenv("DB_URL") + # Set serverSelectionTimeoutMS to 5000 milliseconds (5 second) + self.client = MongoClient(db_url, + tlsCAFile=where(), + tls=True, + tlsAllowInvalidCertificates=True, + serverSelectionTimeoutMS=5000) + self.db_class = Database("test_collection") -# class TestDatabase(unittest.TestCase): -# def setUp(self): -# self.db = Database('test_collection') -# def test_seed(self): -# self.db.seed(10) -# self.assertEqual(self.db.count(), 10) + def test_connection(self): + try: + self.client.admin.command('ismaster') + except ServerSelectionTimeoutError: + self.fail("MongoDB connection failed.") -# def test_reset(self): -# self.db.reset() -# self.assertEqual(self.db.count(), 0) + def test_generate_clone_trooper(self): + n = 5 + result = self.db_class.generate_clone_trooper(n) + self.assertEqual(len(result), n) -# def test_dataframe(self): -# self.db.seed(10) -# df = self.db.dataframe() -# self.assertEqual(len(df), 10) -# def test_html_table(self): -# self.db.seed(10) -# html = self.db.html_table() -# self.assertIsNotNone(html) -# if __name__ == '__main__': -# unittest.main() \ No newline at end of file +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/app/data2.py b/app/data2.py new file mode 100644 index 0000000..1588914 --- /dev/null +++ b/app/data2.py @@ -0,0 +1,163 @@ +# Standard library imports +from datetime import datetime +from os import getenv +import math +import random +import unittest + +from dotenv import load_dotenv +import pandas as pd + +import psycopg2 + +# Typing +from typing import Dict, Iterable, Iterator, List, Tuple + +# List and Etc. + +ranks = ['Private', 'Sergeant', 'Lieutenant', 'Captain', 'Commander'] +rank_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] + +clone_types = ['Standard','Recon', 'Heavy', 'Commando', 'ARC Trooper'] +clone_type_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] + +assigned_weapons = ["DC-15A Blaster Rifle", + "DC-15X Sniper Rifle", + "DC-17m Blaster Rifle", + "DC-15S Blaster Carbine", + "WESTAR-M5 Blaster Rifle"] + +assigned_generals = ['General Mace Windu', "General Plo Koon", 'General Yoda', 'General Kenobi', 'General Skywalker',] +assigned_general_probabilities = [round(random.uniform(0.0,0.1),2), + round(random.uniform(0.1,0.15),2), + round(random.uniform(0.2,0.25),2), + round(random.uniform(0.25,0.3),2), + round(random.uniform(0.3,0.35),2)] + + +class Database: + load_dotenv() + + def __init__(self): + self.conn = psycopg2.connect( + dbname=getenv("DB_NAME"), + user=getenv("DB_USER"), + password=getenv("DB_PASSWORD"), + host=getenv("DB_HOST") + ) + self.cur = self.conn.cursor() + + def create_table(self): + self.cur.execute(""" + CREATE TABLE IF NOT EXISTS clone_troopers ( + ct_id SERIAL PRIMARY KEY, + clone_type VARCHAR(50), + rank VARCHAR(50), + assigned_weapon VARCHAR(50), + health INTEGER, + energy INTEGER, + success_percentage NUMERIC, + assigned_general VARCHAR(50), + check_in_time TIMESTAMP + ) + """) + self.conn.commit() + + def seed(self, amount): + data = self.generate_clone_trooper(amount) + for row in data: + self.cur.execute(""" + INSERT INTO clone_troopers (clone_type, rank, assigned_weapon, health, energy, success_percentage, assigned_general, check_in_time) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, row) + self.conn.commit() + + def reset(self) -> int: + self.cur.execute("DELETE FROM clone_troopers") + self.conn.commit() + return self.cur.rowcount + + def count(self) -> int: + self.cur.execute("SELECT COUNT(*) FROM clone_troopers") + return self.cur.fetchone()[0] + + def dataframe(self) -> pd.DataFrame: + self.cur.execute("SELECT * FROM clone_troopers") + rows = self.cur.fetchall() + columns = [desc[0] for desc in self.cur.description] + return pd.DataFrame(rows, columns=columns) + + def html_table(self) -> str: + df = self.to_dataframe() + return df.to_html() + + # Generate a random clone trooper! + @staticmethod + def generate_clone_trooper(n: int) -> List[Tuple]: + data = [] + for _ in range(n): + clone_type = random.choices(clone_types, clone_type_probabilities)[0] + rank = random.choices(ranks, rank_probabilities)[0] + assigned_weapon = random.choice(assigned_weapons) + health = random.randint(3, 9) + energy = random.randint(2, 6) + assigned_general = random.choices(assigned_generals, assigned_general_probabilities)[0] + success_percentage = round(sum([random.uniform(0, 0.35), math.log10(health)-0.80]), 2) + check_in_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + data.append((clone_type, rank, assigned_weapon, health, energy, success_percentage, assigned_general, check_in_time)) + return data + +class TestDatabase(unittest.TestCase): + def setUp(self): + self.db = Database() # Adjust this to match your setup + + def test_connection(self): + self.assertIsNotNone(self.db.conn) + + def test_seed(self): + self.db.reset() + self.db.seed(10) + self.assertEqual(self.db.count(), 10) + + def test_reset(self): + self.db.reset() + self.db.seed(10) + self.db.reset() + self.assertEqual(self.db.count(), 0) + + def test_count(self): + self.db.reset() + self.db.seed(5) + self.assertEqual(self.db.count(), 5) + + def test_dataframe(self): + self.db.reset() + self.db.seed(5) + df = self.db.dataframe() + self.assertIsInstance(df, DataFrame) + self.assertEqual(len(df), 5) + + def test_html_table(self): + self.db.reset() + self.db.seed(5) + html = self.db.html_table() + self.assertIsInstance(html, str) + self.assertTrue('