Skip to content

Commit

Permalink
Remove all unwanted parameters and replace with env access
Browse files Browse the repository at this point in the history
  • Loading branch information
dormant-user committed Nov 29, 2024
1 parent 9e025fd commit 741732e
Show file tree
Hide file tree
Showing 10 changed files with 98 additions and 179 deletions.
2 changes: 1 addition & 1 deletion docs/README.html
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" /><meta name="generator" content="Docutils 0.19: https://docutils.sourceforge.io/" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" /><meta name="viewport" content="width=device-width, initial-scale=1" />

<title>PyNinja &#8212; PyNinja documentation</title>
<link rel="stylesheet" type="text/css" href="_static/pygments.css" />
Expand Down
140 changes: 49 additions & 91 deletions docs/index.html

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion docs/searchindex.js

Large diffs are not rendered by default.

34 changes: 10 additions & 24 deletions pyninja/features/cpu.py
Original file line number Diff line number Diff line change
@@ -1,42 +1,28 @@
import logging
import subprocess

from pydantic import FilePath

from pyninja.modules import enums, models

LOGGER = logging.getLogger("uvicorn.default")


def _darwin(lib_path: FilePath) -> str:
"""Get processor information for macOS.
Args:
lib_path: Path to the library executable.
"""
command = [lib_path, "-n", "machdep.cpu.brand_string"]
def _darwin() -> str:
"""Get processor information for macOS."""
command = [models.env.processor_lib, "-n", "machdep.cpu.brand_string"]
return subprocess.check_output(command).decode().strip()


def _linux(lib_path: FilePath) -> str:
"""Get processor information for Linux.
Args:
lib_path: Path to the library file.
"""
with open(lib_path) as file:
def _linux() -> str:
"""Get processor information for Linux."""
with open(models.env.processor_lib) as file:
for line in file:
if "model name" in line:
return line.split(":")[1].strip()


def _windows(lib_path: FilePath) -> str:
"""Get processor information for Windows.
Args:
lib_path: Path to the library file.
"""
command = f"{lib_path} cpu get name"
def _windows() -> str:
"""Get processor information for Windows."""
command = f"{models.env.processor_lib} cpu get name"
output = subprocess.check_output(command, shell=True).decode()
return output.strip().split("\n")[1]

Expand All @@ -54,6 +40,6 @@ def get_name() -> str | None:
enums.OperatingSystem.windows: _windows,
}
try:
return os_map[models.OPERATING_SYSTEM](models.env.processor_lib)
return os_map[models.OPERATING_SYSTEM]()
except Exception as error:
LOGGER.error(error)
2 changes: 1 addition & 1 deletion pyninja/features/disks/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,6 @@ def get_all_disks() -> List[Dict[str, str]]:
enums.OperatingSystem.windows: windows.drive_info,
}
try:
return os_map[models.OPERATING_SYSTEM](models.env.disk_lib)
return os_map[models.OPERATING_SYSTEM]()
except Exception as error:
LOGGER.error(error)
9 changes: 3 additions & 6 deletions pyninja/features/disks/linux.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,19 @@
import subprocess
from typing import Dict, List

from pydantic import FilePath
from pyninja.modules import models


def drive_info(lib_path: FilePath) -> List[Dict[str, str]]:
def drive_info() -> List[Dict[str, str]]:
"""Get disks attached to Linux devices.
Args:
lib_path: Returns the library path for disk information.
Returns:
List[Dict[str, str]]:
Returns disks information for Linux distros.
"""
# Using -d to list only physical disks, and filtering out loop devices
result = subprocess.run(
[lib_path, "-o", "NAME,SIZE,TYPE,MODEL,MOUNTPOINT", "-J"],
[models.env.disk_lib, "-o", "NAME,SIZE,TYPE,MODEL,MOUNTPOINT", "-J"],
capture_output=True,
text=True,
)
Expand Down
12 changes: 5 additions & 7 deletions pyninja/features/disks/macOS.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,8 @@
from collections import defaultdict
from typing import Dict, List

from pydantic import FilePath

from pyninja.executors import squire
from pyninja.modules import models

LOGGER = logging.getLogger("uvicorn.default")

Expand Down Expand Up @@ -75,17 +74,16 @@ def parse_diskutil_output(stdout: str) -> List[Dict[str, str]]:
return disks


def drive_info(lib_path: FilePath) -> List[Dict[str, str]]:
def drive_info() -> List[Dict[str, str]]:
"""Get disks attached to macOS devices.
Args:
lib_path: Returns the library path for disk information.
Returns:
List[Dict[str, str]]:
Returns disks information for macOS devices.
"""
result = subprocess.run([lib_path, "info", "-all"], capture_output=True, text=True)
result = subprocess.run(
[models.env.disk_lib, "info", "-all"], capture_output=True, text=True
)
disks = parse_diskutil_output(result.stdout)
device_ids = defaultdict(list)
physical_disks = []
Expand Down
36 changes: 12 additions & 24 deletions pyninja/features/disks/windows.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,8 @@
import subprocess
from typing import Dict, List, Tuple

from pydantic import FilePath

from pyninja.executors import squire
from pyninja.modules import models

LOGGER = logging.getLogger("uvicorn.default")

Expand All @@ -31,19 +30,17 @@ def reformat_windows(data: Dict[str, str | int | float]) -> Dict[str, str]:
return data


def get_drives(lib_path: FilePath) -> List[Dict[str, str]]:
def get_drives() -> List[Dict[str, str]]:
"""Get physical drives connected to a Windows machine.
Args:
lib_path: Library path for disk information.
Returns:
List[Dict[str, str]]:
Returns the formatted data for all the drives as a list of key-value pairs.
"""
# noinspection LongLine
ps_command = "Get-CimInstance Win32_DiskDrive | Select-Object Caption, DeviceID, Model, Partitions, Size | ConvertTo-Json" # noqa: E501
result = subprocess.run(
[lib_path, "-Command", ps_command], capture_output=True, text=True
[models.env.disk_lib, "-Command", ps_command], capture_output=True, text=True
)
disks_info = json.loads(result.stdout)
if isinstance(disks_info, list):
Expand All @@ -65,18 +62,15 @@ def clean_ansi_escape_sequences(text: str) -> str:
return ansi_escape.sub("", text)


def get_physical_disks_and_partitions(lib_path: FilePath) -> List[Tuple[str, str, str]]:
def get_physical_disks_and_partitions() -> List[Tuple[str, str, str]]:
"""Powershell Core command to get physical disks and their partitions with drive letters (mount points).
Args:
lib_path: Library path for disk information.
Returns:
List[Tuple[str, str, str]]:
List of tuples with disk_number, partition_number, mount_point.
"""
command_ps = [
lib_path,
models.env.disk_lib,
"-Command",
"""
Get-PhysicalDisk | ForEach-Object {
Expand All @@ -100,7 +94,7 @@ def get_physical_disks_and_partitions(lib_path: FilePath) -> List[Tuple[str, str
)

if result.stderr:
print("Error:", result.stderr)
LOGGER.error(result.stderr)
return []

# Clean the output to remove ANSI escape sequences
Expand All @@ -126,17 +120,14 @@ def get_physical_disks_and_partitions(lib_path: FilePath) -> List[Tuple[str, str
return disks_and_partitions


def get_disk_usage(lib_path: FilePath) -> Dict[str, List[str]]:
def get_disk_usage() -> Dict[str, List[str]]:
"""Get all physical disks and their partitions with mount points.
Args:
lib_path: Library path for disk information.
Returns:
Dict[str, List[str]]:
Returns a dictionary of DeviceID as key and mount paths as value.
"""
disks_and_partitions = get_physical_disks_and_partitions(lib_path)
disks_and_partitions = get_physical_disks_and_partitions()

if not disks_and_partitions:
LOGGER.error("No disks or partitions found.")
Expand All @@ -151,18 +142,15 @@ def get_disk_usage(lib_path: FilePath) -> Dict[str, List[str]]:
return output_data


def drive_info(lib_path: FilePath) -> List[Dict[str, str]]:
def drive_info() -> List[Dict[str, str]]:
"""Get disks attached to Windows devices.
Args:
lib_path: Library path for disk information.
Returns:
List[Dict[str, str]]:
Returns disks information for Windows machines.
"""
data = get_drives(lib_path)
usage = get_disk_usage(lib_path)
data = get_drives()
usage = get_disk_usage()
for item in data:
device_id = item["ID"]
item.pop("ID")
Expand Down
25 changes: 7 additions & 18 deletions pyninja/features/gpu.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,20 @@
import subprocess
from typing import Dict, List, Optional

from pydantic import FilePath

from pyninja.modules import models

LOGGER = logging.getLogger("uvicorn.default")


def _darwin(lib_path) -> Optional[List[Dict[str, str]]]:
def _darwin() -> Optional[List[Dict[str, str]]]:
"""Get GPU model and vendor information for Linux operating system.
Args:
lib_path: Library path to get GPU information.
Returns:
List[Dict[str, str]]:
Returns a list of GPU model and vendor information.
"""
result = subprocess.run(
[lib_path, "SPDisplaysDataType", "-json"],
[models.env.gpu_lib, "SPDisplaysDataType", "-json"],
capture_output=True,
text=True,
)
Expand All @@ -45,18 +40,15 @@ def _darwin(lib_path) -> Optional[List[Dict[str, str]]]:
return gpu_info


def _linux(lib_path) -> Optional[List[Dict[str, str]]]:
def _linux() -> Optional[List[Dict[str, str]]]:
"""Get GPU model and vendor information for Linux operating system.
Args:
lib_path: Library path to get GPU information.
Returns:
List[Dict[str, str]]:
Returns a list of GPU model and vendor information.
"""
result = subprocess.run(
[lib_path],
[models.env.gpu_lib],
capture_output=True,
text=True,
)
Expand All @@ -78,19 +70,16 @@ def _linux(lib_path) -> Optional[List[Dict[str, str]]]:
return gpu_info


def _windows(lib_path: FilePath) -> Optional[List[Dict[str, str]]]:
def _windows() -> Optional[List[Dict[str, str]]]:
"""Get GPU model and vendor information for Windows operating system.
Args:
lib_path: Library path to get GPU information.
Returns:
List[Dict[str, str]]:
Returns a list of GPU model and vendor information.
"""
result = subprocess.run(
[
lib_path,
models.env.gpu_lib,
"path",
"win32_videocontroller",
"get",
Expand Down Expand Up @@ -129,6 +118,6 @@ def get_names() -> List[Dict[str, str]]:
"""Get list of GPU model and vendor information based on the operating system."""
fn_map = dict(linux=_linux, darwin=_darwin, windows=_windows)
try:
return fn_map[models.OPERATING_SYSTEM](models.env.gpu_lib)
return fn_map[models.OPERATING_SYSTEM]()
except (subprocess.SubprocessError, FileNotFoundError) as error:
LOGGER.debug(error)
15 changes: 9 additions & 6 deletions pyninja/features/operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,13 +173,14 @@ def get_service_pid_linux(service_name: str) -> Optional[int]:
"""
try:
output = subprocess.check_output(
["systemctl", "show", service_name, "--property=MainPID"], text=True
[models.env.service_lib, "show", service_name, "--property=MainPID"],
text=True,
)
for line in output.splitlines():
if line.startswith("MainPID="):
return int(line.split("=")[1].strip())
except subprocess.CalledProcessError as error:
LOGGER.error("%s - %s", error.returncode, error.stderr)
LOGGER.debug("%s - %s", error.returncode, error.stderr)


def get_service_pid_macos(service_name: str) -> Optional[int]:
Expand All @@ -193,12 +194,12 @@ def get_service_pid_macos(service_name: str) -> Optional[int]:
Returns the PID of the service.
"""
try:
output = subprocess.check_output(["launchctl", "list"], text=True)
output = subprocess.check_output([models.env.service_lib, "list"], text=True)
for line in output.splitlines():
if service_name in line:
return int(line.split()[0]) # Assuming PID is the first column
except subprocess.CalledProcessError as error:
LOGGER.error("%s - %s", error.returncode, error.stderr)
LOGGER.debug("%s - %s", error.returncode, error.stderr)


def get_service_pid_windows(service_name: str) -> Optional[int]:
Expand All @@ -212,9 +213,11 @@ def get_service_pid_windows(service_name: str) -> Optional[int]:
Returns the PID of the service.
"""
try:
output = subprocess.check_output(["sc", "query", service_name], text=True)
output = subprocess.check_output(
[models.env.service_lib, "query", service_name], text=True
)
for line in output.splitlines():
if "PID" in line:
return int(line.split(":")[1].strip())
except subprocess.CalledProcessError as error:
LOGGER.error("%s - %s", error.returncode, error.stderr)
LOGGER.debug("%s - %s", error.returncode, error.stderr)

0 comments on commit 741732e

Please sign in to comment.