-
Notifications
You must be signed in to change notification settings - Fork 0
/
environment.py
64 lines (55 loc) · 2.23 KB
/
environment.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import asyncio
from pathlib import Path
import aiohttp
import logging
class Environment:
has_internet = False
@staticmethod
def get_project_root(relative_path='.git'):
current = Path(__file__) # Start from the current file location
while current != current.parent:
if current.joinpath(relative_path).exists(): # Check if the marker exists at this level
return current
current = current.parent
return current # Fallback to the current directory if nothing is found
@staticmethod
async def check_internet():
url = 'http://www.google.com/'
timeout = aiohttp.ClientTimeout(total=5)
try:
logging.debug("Attempting to reach Google...")
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as response:
if response.status == 200:
Environment.has_internet = True
logging.debug("Internet connection successful.")
else:
Environment.has_internet = False
logging.debug(f"Received non-success status code: {response.status}")
except aiohttp.ClientError as ce:
Environment.has_internet = False
logging.debug(f"Client error: {ce}")
except asyncio.TimeoutError as te:
Environment.has_internet = False
logging.debug(f"Timeout error: {te}")
return Environment.has_internet
@staticmethod
async def start_internet_check(interval=60):
while True:
await Environment.check_internet()
await asyncio.sleep(interval)
# Setup logging
# Usage
async def main():
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
# Start checking for internet connectivity
await asyncio.create_task(Environment.start_internet_check(interval=60))
# Example of other asynchronous tasks
await asyncio.sleep(10)
if Environment.has_internet:
print("Connected to the internet.")
else:
print("Not connected to the internet.")
if __name__ == '__main__':
# Run the main coroutine
asyncio.run(main())