From 70af79e4db597cebcfff31b92c1feaa124f59bee Mon Sep 17 00:00:00 2001 From: Florian Sylvain Date: Fri, 28 Mar 2025 00:23:46 +0100 Subject: [PATCH] refactor: mv weather & draw code into their own modules --- constants.py | 2 +- draw.py | 128 +++++++++++++++++++++++++++++++++++++++ main.py | 167 +++++---------------------------------------------- weather.py | 54 +++++++++++++++++ 4 files changed, 199 insertions(+), 152 deletions(-) create mode 100644 draw.py create mode 100644 weather.py diff --git a/constants.py b/constants.py index 8eb1825..88d5041 100644 --- a/constants.py +++ b/constants.py @@ -9,4 +9,4 @@ OPENMETEO_HOURLY_PARAMS = [ "visibility", "precipitation", "cloud_cover", "pressure_msl", "surface_pressure", "wind_direction_10m", "shortwave_radiation", "direct_radiation", "diffuse_radiation", "dew_point_2m" -] \ No newline at end of file +] diff --git a/draw.py b/draw.py new file mode 100644 index 0000000..d86d6ce --- /dev/null +++ b/draw.py @@ -0,0 +1,128 @@ +""" +This module handles the generation and display of lottery draws. +""" + +import os +import time +import hashlib +import secrets +import random +import socket +import platform +import uuid +import threading +import datetime +import multiprocessing +import psutil +from progress.bar import Bar + +from constants import MAX_NUMBER, MAX_STAR, NUMBER_OF_NUMBERS, NUMBER_OF_STARS + + +class Draw: + """Handles the generation and display of lottery draws""" + + def __init__(self, weather_service): + self.weather_service = weather_service + + def get_static_entropy_pool(self): + """Creates a base entropy pool with sources that don't change rapidly""" + weather_entropy = self.weather_service.get_weather_entropy() + return [ + str(os.urandom(32)), + str(secrets.token_bytes(32)), + socket.gethostname(), + str(platform.system_alias(platform.system(), + platform.release(), platform.version())), + str(uuid.getnode()), + str(multiprocessing.cpu_count()), + "".join([str(v) for v in psutil.disk_partitions()]), + str(hash(frozenset(os.environ.items()))), + weather_entropy, + ] + + def get_dynamic_entropy_pool(self): + """Retrieves dynamic data to add entropy""" + return [ + str(time.time()), + hashlib.sha256(str(time.perf_counter()).encode()).hexdigest(), + str(os.getpid()), + str(psutil.cpu_percent(interval=0.01)), + str(psutil.virtual_memory().percent), + str(psutil.disk_usage('/').percent), + str(random.getrandbits(256)), + str(datetime.datetime.now().microsecond), + str(threading.active_count()), + str(sum(psutil.cpu_times())), + str(psutil.net_io_counters().bytes_sent if hasattr( + psutil, 'net_io_counters') else 0), + str(id({})), + ] + + def generate_seed(self, base_pool=None): + """Generates a random seed by combining different entropy sources""" + entropy_sources = base_pool.copy() if base_pool else [] + entropy_sources.extend(self.get_dynamic_entropy_pool()) + random.shuffle(entropy_sources) + + entropy_str = "".join(entropy_sources) + intermediate_hash = hashlib.sha512(entropy_str.encode()).digest() + intermediate_hash = hashlib.blake2b(intermediate_hash).digest() + + final_hash = hashlib.sha256(intermediate_hash).hexdigest() + return int(final_hash, 16) + + def make_draw(self, base_pool=None): + """Generates a draw with N numbers and M stars using a random seed""" + seed = self.generate_seed(base_pool) + random.seed(seed) + numbers = random.sample(range(1, MAX_NUMBER + 1), NUMBER_OF_NUMBERS) + stars = random.sample(range(1, MAX_STAR + 1), NUMBER_OF_STARS) + return { + "seed": seed, + "numbers": sorted(numbers), + "stars": sorted(stars) + } + + def generate_draws(self, num_draws): + """Generates multiple draws sequentially""" + base_pool = self.get_static_entropy_pool() + progress_bar = Bar('Progress', max=num_draws, suffix='%(percent)d%%') + + draws = [] + for _ in range(num_draws): + draws.append(self.make_draw(base_pool)) + time.sleep(0.01) + progress_bar.next() + + progress_bar.finish() + return draws + + def display_draw(self, draw, index=None, title="DRAW"): + """Displays a draw in a formatted way""" + print(f"\n===== {title} =====") + if index is not None: + print(f"Draw #{index + 1}") + print(f"Numbers: {draw['numbers']}") + print(f"Stars: {draw['stars']}") + print(f"Seed used: {draw['seed']}") + + def display_additional_draws(self, draws, displayed_draws): + """Displays additional random draws at the user's request.""" + while len(displayed_draws) < len(draws): + display_another = input( + "\nDo you want to see another random draw? (y/n): ").lower().strip() + if display_another not in ['o', 'oui', 'y', 'yes', '']: + break + + available_indices = [i for i in range( + len(draws)) if i not in displayed_draws] + if not available_indices: + print("All draws have already been displayed!") + break + + new_index = random.choice(available_indices) + new_draw = draws[new_index] + displayed_draws.add(new_index) + + self.display_draw(new_draw, new_index, "ANOTHER RANDOM DRAW") diff --git a/main.py b/main.py index f0394e5..37280ef 100644 --- a/main.py +++ b/main.py @@ -2,166 +2,31 @@ This script generates random draws for EuroMillions using various entropy sources. """ -import os -import time -import hashlib -import secrets import random -import socket -import platform -import psutil -import uuid -import threading -import datetime -import multiprocessing -from progress.bar import Bar -import openmeteo_requests -import requests_cache import numpy as _ -from retry_requests import retry -from constants import * -def _fetch_weather_api(params): - """Internal function to make Open-Meteo API requests""" - cache_session = requests_cache.CachedSession('.cache', expire_after=60) - retry_session = retry(cache_session, retries=2, backoff_factor=0.2) - openmeteo = openmeteo_requests.Client(session=retry_session) - return openmeteo.weather_api(OPENMETEO_API_URL, params=params) +from draw import Draw +from weather import Weather +from constants import NUMBER_OF_DRAWS -def _get_weather_data(): - """Internal function that retrieves weather data from API""" - params = { - "latitude": random.uniform(-70, 70), - "longitude": random.uniform(-180, 180), - "hourly": random.sample(OPENMETEO_HOURLY_PARAMS, random.randint(3, 6)), - "timezone": "auto" - } - responses = _fetch_weather_api(params) - entropy_values = [] - for i in range(len(params["hourly"])): - var_values = responses[0].Hourly().Variables(i).ValuesAsNumpy() - entropy_values.extend(var_values.tolist()) +def main(): + """Main function that orchestrates the draw generation process""" + weather = Weather() + draw_generator = Draw(weather) - weather_str = "".join([str(v) for v in entropy_values]) - return hashlib.sha256(weather_str.encode()).hexdigest() - -def get_weather_entropy(): - """Retrieves weather data as additional source of entropy""" - try: - return _get_weather_data() - except Exception as e: - print(f"Error retrieving weather data: {e}") - fallback = f"weather_fallback_{time.time()}_{random.getrandbits(64)}" - return hashlib.sha256(fallback.encode()).hexdigest() - -def get_static_entropy_pool(): - """Creates a base entropy pool with sources that don't change rapidly""" - weather_entropy = get_weather_entropy() - return [ - str(os.urandom(32)), - str(secrets.token_bytes(32)), - socket.gethostname(), - str(platform.system_alias(platform.system(), platform.release(), platform.version())), - str(uuid.getnode()), - str(multiprocessing.cpu_count()), - "".join([str(v) for v in psutil.disk_partitions()]), - str(hash(frozenset(os.environ.items()))), - weather_entropy, - ] - -def get_dynamic_entropy_pool(): - """Retrieves dynamic data to add entropy""" - return [ - str(time.time()), - hashlib.sha256(str(time.perf_counter()).encode()).hexdigest(), - str(os.getpid()), - str(psutil.cpu_percent(interval=0.01)), - str(psutil.virtual_memory().percent), - str(psutil.disk_usage('/').percent), - str(random.getrandbits(256)), - str(datetime.datetime.now().microsecond), - str(threading.active_count()), - str(sum(psutil.cpu_times())), - str(psutil.net_io_counters().bytes_sent if hasattr(psutil, 'net_io_counters') else 0), - str(id({})), - ] - -def generate_seed(base_pool=None): - """Generates a random seed by combining different entropy sources""" - entropy_sources = base_pool.copy() if base_pool else [] - entropy_sources.extend(get_dynamic_entropy_pool()) - random.shuffle(entropy_sources) - - entropy_str = "".join(entropy_sources) - intermediate_hash = hashlib.sha512(entropy_str.encode()).digest() - intermediate_hash = hashlib.blake2b(intermediate_hash).digest() - - final_hash = hashlib.sha256(intermediate_hash).hexdigest() - return int(final_hash, 16) - -def make_draw(base_pool=None): - """Generates a draw with N numbers and M stars using a random seed""" - seed = generate_seed(base_pool) - random.seed(seed) - numbers = random.sample(range(1, MAX_NUMBER + 1), NUMBER_OF_NUMBERS) - stars = random.sample(range(1, MAX_STAR + 1), NUMBER_OF_STARS) - return { - "seed": seed, - "numbers": sorted(numbers), - "stars": sorted(stars) - } - -def generate_draws(num_draws): - """Generates multiple draws sequentially""" - base_pool = get_static_entropy_pool() - bar = Bar('Progress', max=num_draws, suffix='%(percent)d%%') - - draws = [] - for _ in range(num_draws): - draws.append(make_draw(base_pool)) - time.sleep(0.01) - bar.next() - - bar.finish() - return draws - -def display_draw(draw, index=None, title="DRAW"): - """Displays a draw in a formatted way""" - print(f"\n===== {title} =====") - if index is not None: - print(f"Draw #{index + 1}") - print(f"Numbers: {draw['numbers']}") - print(f"Stars: {draw['stars']}") - print(f"Seed used: {draw['seed']}") - -def display_additional_draws(draws, displayed_draws): - """Displays additional random draws at the user's request.""" - while len(displayed_draws) < len(draws): - display_another = input("\nDo you want to see another random draw? (y/n): ").lower().strip() - if display_another not in ['o', 'oui', 'y', 'yes', '']: - break - - available_indices = [i for i in range(len(draws)) if i not in displayed_draws] - if not available_indices: - print("All draws have already been displayed!") - break - - new_index = random.choice(available_indices) - new_draw = draws[new_index] - displayed_draws.add(new_index) - - display_draw(new_draw, new_index, "ANOTHER RANDOM DRAW") - -if __name__ == "__main__": print(f"Generating {NUMBER_OF_DRAWS} different draws...") - draws = generate_draws(NUMBER_OF_DRAWS) + draws = draw_generator.generate_draws(NUMBER_OF_DRAWS) - base_pool = get_static_entropy_pool() - random.seed(generate_seed(base_pool)) + base_pool = draw_generator.get_static_entropy_pool() + random.seed(draw_generator.generate_seed(base_pool)) chosen_draw = random.choice(draws) displayed_draws = {draws.index(chosen_draw)} - display_draw(chosen_draw, title="FINAL RESULT") + draw_generator.display_draw(chosen_draw, title="FINAL RESULT") print("Draw selected from among the", NUMBER_OF_DRAWS, "generated") - display_additional_draws(draws, displayed_draws) + draw_generator.display_additional_draws(draws, displayed_draws) + + +if __name__ == "__main__": + main() diff --git a/weather.py b/weather.py new file mode 100644 index 0000000..8887ec0 --- /dev/null +++ b/weather.py @@ -0,0 +1,54 @@ +""" +Weather module for entropy generation +""" + +import time +import hashlib +import random +import openmeteo_requests +import requests_cache +from retry_requests import retry + +from constants import OPENMETEO_API_URL, OPENMETEO_HOURLY_PARAMS + + +class Weather: + """Handles weather data retrieval for entropy generation""" + + def __init__(self): + self.api_url = OPENMETEO_API_URL + self.hourly_params = OPENMETEO_HOURLY_PARAMS + + def _fetch_weather_api(self, params): + """Internal function to make Open-Meteo API requests""" + cache_session = requests_cache.CachedSession('.cache', expire_after=60) + retry_session = retry(cache_session, retries=2, backoff_factor=0.2) + openmeteo = openmeteo_requests.Client(session=retry_session) + return openmeteo.weather_api(self.api_url, params=params) + + def _get_weather_data(self): + """Internal function that retrieves weather data from API""" + params = { + "latitude": random.uniform(-70, 70), + "longitude": random.uniform(-180, 180), + "hourly": random.sample(self.hourly_params, random.randint(3, 6)), + "timezone": "auto" + } + + responses = self._fetch_weather_api(params) + entropy_values = [] + for i in range(len(params["hourly"])): + var_values = responses[0].Hourly().Variables(i).ValuesAsNumpy() + entropy_values.extend(var_values.tolist()) + + weather_str = "".join([str(v) for v in entropy_values]) + return hashlib.sha256(weather_str.encode()).hexdigest() + + def get_weather_entropy(self): + """Retrieves weather data as additional source of entropy""" + try: + return self._get_weather_data() + except Exception as e: + print(f"Error retrieving weather data: {e}") + fallback = f"weather_fallback_{time.time()}_{random.getrandbits(64)}" + return hashlib.sha256(fallback.encode()).hexdigest()