ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[CLEANUP UN-USED FILES SUCH AS RUN_CPU decorator + other files]

Browse Source
pull/634/head
Your Name 2 months ago
parent e0ca51b560
commit f4cf551c23
4 changed files with 57 additions and 150 deletions
Show Stats Download Patch File Download Diff File

28
swarms/telemetry/bootup.py
Unescape View File

@ -9,18 +9,22 @@ from swarms.utils.disable_logging import disable_logging
def bootup(): def bootup():
"""Bootup swarms""" """Bootup swarms"""
logging.disable(logging.CRITICAL) try:
os.environ["WANDB_SILENT"] = "true" logging.disable(logging.CRITICAL)
os.environ["WANDB_SILENT"] = "true"
# Auto set workspace directory # Auto set workspace directory
workspace_dir = os.path.join(os.getcwd(), "agent_workspace") workspace_dir = os.path.join(os.getcwd(), "agent_workspace")
if not os.path.exists(workspace_dir): if not os.path.exists(workspace_dir):
os.makedirs(workspace_dir) os.makedirs(workspace_dir, exist_ok=True)
os.environ["WORKSPACE_DIR"] = workspace_dir os.environ["WORKSPACE_DIR"] = workspace_dir
warnings.filterwarnings("ignore", category=DeprecationWarning) warnings.filterwarnings("ignore", category=DeprecationWarning)
# Use ThreadPoolExecutor to run disable_logging and auto_update concurrently # Use ThreadPoolExecutor to run disable_logging and auto_update concurrently
with ThreadPoolExecutor(max_workers=2) as executor: with ThreadPoolExecutor(max_workers=2) as executor:
executor.submit(disable_logging) executor.submit(disable_logging)
executor.submit(auto_update) executor.submit(auto_update)
except Exception as e:
print(f"An error occurred: {str(e)}")
raise

51
swarms/telemetry/check_update.py
Unescape View File

@ -4,10 +4,22 @@ import sys
import pkg_resources import pkg_resources
import requests import requests
from packaging import version from packaging import version
from swarms.utils.loguru_logger import initialize_logger
logger = initialize_logger("check-update")
# borrowed from: https://stackoverflow.com/a/1051266/656011 # borrowed from: https://stackoverflow.com/a/1051266/656011
def check_for_package(package: str) -> bool: def check_for_package(package: str) -> bool:
"""
Checks if a package is installed and available for import.
Args:
package (str): The name of the package to check.
Returns:
bool: True if the package is installed and can be imported, False otherwise.
"""
if package in sys.modules: if package in sys.modules:
return True return True
elif (spec := importlib.util.find_spec(package)) is not None: elif (spec := importlib.util.find_spec(package)) is not None:
@ -19,24 +31,43 @@ def check_for_package(package: str) -> bool:
return True return True
except ImportError: except ImportError:
logger.error(f"Failed to import {package}")
return False return False
else: else:
logger.info(f"{package} not found")
return False return False
def check_for_update() -> bool: def check_for_update() -> bool:
"""Check for updates """
Checks if there is an update available for the swarms package.
Returns: Returns:
BOOL: Flag to indicate if there is an update bool: True if an update is available, False otherwise.
""" """
# Fetch the latest version from the PyPI API try:
response = requests.get("https://pypi.org/pypi/swarms/json") # Fetch the latest version from the PyPI API
latest_version = response.json()["info"]["version"] response = requests.get("https://pypi.org/pypi/swarms/json")
response.raise_for_status() # Raises an HTTPError if the response status code is 4XX/5XX
latest_version = response.json()["info"]["version"]
# Get the current version using pkg_resources # Get the current version using pkg_resources
current_version = pkg_resources.get_distribution("swarms").version current_version = pkg_resources.get_distribution(
"swarms"
).version
return version.parse(latest_version) > version.parse( if version.parse(latest_version) > version.parse(
current_version current_version
) ):
logger.info(
f"Update available: {latest_version} > {current_version}"
)
return True
else:
logger.info(
f"No update available: {latest_version} <= {current_version}"
)
return False
except requests.exceptions.RequestException as e:
logger.error(f"Failed to check for update: {e}")
return False

0
swarms/utils/profile_func_2.py
Unescape View File

128
swarms/utils/run_on_cpu.py
Unescape View File

@ -1,128 +0,0 @@
import os
import psutil
from typing import Callable, Any
import functools
from swarms.utils.loguru_logger import initialize_logger
logger = initialize_logger(log_folder="run_on_cpu")
def run_on_cpu(func: Callable) -> Callable:
"""
Decorator that ensures the function runs on all available CPU cores,
maximizing CPU and memory usage to execute the function as quickly as possible.
This decorator sets the CPU affinity of the current process to all available CPU cores
before executing the function. After the function completes, the original CPU affinity is restored.
Args:
func (Callable): The function to be executed.
Returns:
Callable: The wrapped function with CPU affinity settings applied.
Raises:
RuntimeError: If the CPU affinity cannot be set or restored.
"""
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
# Get the current process
process = psutil.Process(os.getpid())
# Check if the platform supports cpu_affinity
if not hasattr(process, "cpu_affinity"):
logger.warning(
"CPU affinity is not supported on this platform. Executing function without setting CPU affinity."
)
return func(*args, **kwargs)
# Save the original CPU affinity
original_affinity = process.cpu_affinity()
logger.info(f"Original CPU affinity: {original_affinity}")
try:
# Set the CPU affinity to all available CPU cores
all_cpus = list(range(os.cpu_count()))
process.cpu_affinity(all_cpus)
logger.info(f"Set CPU affinity to: {all_cpus}")
# Set process priority to high
try:
process.nice(psutil.HIGH_PRIORITY_CLASS)
logger.info("Set process priority to high.")
except AttributeError:
logger.warning(
"Setting process priority is not supported on this platform."
)
# Pre-allocate memory by creating a large array (optional step)
memory_size = int(
psutil.virtual_memory().available * 0.9
) # 90% of available memory
try:
logger.info(
f"Pre-allocating memory: {memory_size} bytes"
)
_ = bytearray(memory_size)
except MemoryError:
logger.error(
"Failed to pre-allocate memory, continuing without pre-allocation."
)
# Run the function
result = func(*args, **kwargs)
except psutil.AccessDenied as e:
logger.error(
"Access denied while setting CPU affinity",
exc_info=True,
)
raise RuntimeError(
"Access denied while setting CPU affinity"
) from e
except psutil.NoSuchProcess as e:
logger.error("Process does not exist", exc_info=True)
raise RuntimeError("Process does not exist") from e
except Exception as e:
logger.error(
"An error occurred during function execution",
exc_info=True,
)
raise RuntimeError(
"An error occurred during function execution"
) from e
finally:
# Restore the original CPU affinity
try:
process.cpu_affinity(original_affinity)
logger.info(
f"Restored original CPU affinity: {original_affinity}"
)
except Exception as e:
logger.error(
"Failed to restore CPU affinity", exc_info=True
)
raise RuntimeError(
"Failed to restore CPU affinity"
) from e
return result
return wrapper
# # Example usage of the decorator
# @run_on_cpu
# def compute_heavy_task() -> None:
# # An example task that is CPU and memory intensive
# data = [i**2 for i in range(100000000)]
# sum(data)
# print("Task completed.")
# compute_heavy_task()
Write Preview
Loading…
Cancel
Save