refactor: mv weather & draw code into their own modules

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")

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):
+from draw import Draw
-    """Internal function to make Open-Meteo API requests"""
+from weather import Weather
-    cache_session = requests_cache.CachedSession('.cache', expire_after=60)
+from constants import NUMBER_OF_DRAWS
-    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)
 
-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)
+def main():
-    entropy_values = []
+    """Main function that orchestrates the draw generation process"""
-    for i in range(len(params["hourly"])):
+    weather = Weather()
-        var_values = responses[0].Hourly().Variables(i).ValuesAsNumpy()
+    draw_generator = Draw(weather)
-        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():
-    """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()
+    base_pool = draw_generator.get_static_entropy_pool()
-    random.seed(generate_seed(base_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()

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()