If you have time to implement your suggestion, you can submit pull requests to ghostbsd/software-station
Your idea might look something like this:
import asyncio
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
from subprocess import Popen, PIPE
import socket
import requests
import shlex
async def run_command(cmd):
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
return stdout.decode(), stderr.decode()
def fetch_package_data(package_name):
# Function to fetch data for a single package
pass
def get_all_package_data(package_list):
with ThreadPoolExecutor() as executor:
results = list(executor.map(fetch_package_data, package_list))
return pd.DataFrame(results)
async def network_stat():
cmd = "netstat -rn | grep default"
stdout, _ = await run_command(cmd)
return "UP" if stdout else "DOWN"
async def repo_online():
cmd = "pkg -vv | grep -B 1 'enabled.*yes' | grep url"
stdout, _ = await run_command(cmd)
output = stdout.strip()
parts = list(filter(None, output.split('/')))
if len(parts) < 2:
return False
server = parts[1]
try:
with socket.create_connection((server, 80), timeout=5):
return True
except OSError:
return False
async def repository_is_syncing():
cmd = 'pkg -vv | grep -B 1 "enabled.*yes" | grep url'
stdout, _ = await run_command(cmd)
output = stdout.strip()
parts = output.split('"')
if len(parts) < 2:
return False
pkg_url = parts[1]
syncing_url = f"{pkg_url}/.syncing"
try:
response = await asyncio.to_thread(requests.get, syncing_url)
return response.status_code == 200
except requests.RequestException:
return False
async def sync_with_repository():
cmd = "yes | pkg update -f"
stdout, _ = await run_command(cmd)
if "Newer FreeBSD version" in stdout:
return "UPGRADE"
return "SYNCED" if stdout else "FAILED"
async def start_update_station():
cmd = "update-station check-now"
proc = await asyncio.create_subprocess_shell(cmd)
await proc.wait()
return proc.returncode == 0
async def get_available_packages():
cmd = "pkg rquery -U '%o<EOS>%n<EOS>%v<EOS>%sh<EOS>%c<EOS>%e<EOL>'"
stdout, _ = await run_command(cmd)
raw_data = stdout.strip().split("<EOL>\n")
data = [line.split("<EOS>") for line in raw_data]
columns = ["origin", "name", "version", "size", "comment", "description"]
return pd.DataFrame(data, columns=columns)
async def get_installed_packages():
cmd = "pkg query '%o<EOS>%n<EOS>%v<EOS>%sh<EOS>%c<EOS>%e<EOL>'"
stdout, _ = await run_command(cmd)
raw_data = stdout.strip().split("<EOL>\n")
data = [line.split("<EOS>") for line in raw_data]
columns = ["origin", "name", "version", "size", "comment", "description"]
return pd.DataFrame(data, columns=columns)
def filter_installed_packages(available_df, installed_df):
installed_names = set(installed_df["name"])
return available_df[~available_df["name"].isin(installed_names)]
def packages_to_install(available_df, selected_packages):
return available_df[available_df["name"].isin(selected_packages)]
async def execute_pkg_command(cmd):
stdout, _ = await run_command(cmd)
return stdout
async def delete_packages(pkg):
return await execute_pkg_command(f"pkg delete -y {shlex.quote(pkg)}")
async def fetch_packages(pkg):
return await execute_pkg_command(f"pkg fetch -y {shlex.quote(pkg)}")
async def install_packages(pkg):
return await execute_pkg_command(f"pkg install -y {shlex.quote(pkg)}")
async def get_pkg_to_install_output(packages):
return await execute_pkg_command(f"pkg install -n {shlex.quote(packages)}")
async def get_pkg_to_remove_output(packages):
return await execute_pkg_command(f"pkg delete -n {shlex.quote(packages)}")
async def get_pkg_changes_data(remove_list, install_list):
install_pkg_list = (await get_pkg_to_install_output(" ".join(install_list))).splitlines() if install_list else []
remove_pkg_list = (await get_pkg_to_remove_output(" ".join(remove_list))).splitlines() if remove_list else []
pkg_data = {"remove": [], "upgrade": [], "install": [], "reinstall": []}
stop = False
for line in remove_pkg_list:
if "REMOVED:" in line:
stop = True
elif stop and not line.strip():
break
elif stop:
pkg_data["remove"].append(line.strip())
stop = False
for line in install_pkg_list:
if "UPGRADED:" in line:
stop = True
elif stop and not line.strip():
break
elif stop:
pkg_data["upgrade"].append(line.strip())
stop = False
for line in install_pkg_list:
if "INSTALLED:" in line:
stop = True
elif stop and not line.strip():
break
elif stop:
pkg_data["install"].append(line.strip())
stop = False
for line in install_pkg_list:
if "REINSTALLED:" in line:
stop = True
elif stop and not line.strip():
break
elif stop:
pkg_data["reinstall"].append(line.strip())
return pkg_data
