add argparser mod

This commit is contained in:
2026-05-04 00:13:18 +08:00
parent dc2b21a500
commit 9ec1cf4fdd
11 changed files with 4260 additions and 34 deletions

View File

@@ -3,9 +3,23 @@ import re
import sys import sys
import pwd import pwd
import xml import xml
import yaml
import types
import syslog import syslog
import inspect
import argparse
import sunhpc.util import sunhpc.util
from io import StringIO
from typing import Dict
from sunhpc.logs import Logs
from sunhpc.disks import DiskInfo
from sunhpc.configs import Config
from sunhpc.network import Network
from sunhpc.memory import MemoryInfo
from sunhpc.output import Output
from xml.sax import saxutils from xml.sax import saxutils
from xml.sax import handler from xml.sax import handler
from xml.sax import make_parser from xml.sax import make_parser
@@ -313,20 +327,32 @@ class Command:
def __init__(self, database): def __init__(self, database):
self.db = database self.db = database
self.conf = Config('/etc/sunhpc/sunhpc.yaml')
self.text = '' self.text = ''
self.output = [] self.output = []
self.net = Network()
self.mem = MemoryInfo()
self.disk = DiskInfo()
self.fmt = Output()
self.os = os.uname()[0].lower() self.os = os.uname()[0].lower()
self.arch = os.uname()[4] self.arch = os.uname()[4]
self.user = pwd.getpwuid(os.getuid()).pw_name self.user = pwd.getpwuid(os.getuid()).pw_name
self.parser = argparse.ArgumentParser()
self.subcommand = 'Sub Command'
self.subdesc = None
self.epilog = None
self._args = None self._args = None
self._params = None self._params = None
if 'SUNHPCDEBUG' in os.environ: if 'SUNHPCDEBUG' in os.environ:
self._debug = True self._debug = True
self.log = Logs(name='Sunhpc', level=0, show_time=True)
else: else:
self._debug = False self._debug = False
self.log = Logs(name='Sunhpc', show_time=True)
def debug(self): def debug(self):
"""Return True if debug mode is enabled""" """Return True if debug mode is enabled"""
@@ -366,6 +392,55 @@ class Command:
else: else:
return 'no' return 'no'
def dtons(self, data: dict) -> types.SimpleNamespace:
'''递归将字典转换为 SimpleNamespace'''
if isinstance(data, dict):
for key, val in data.items():
clean_key = key.replace(' ','_')
data[clean_key] = self.dtons(val)
return types.SimpleNamespace(**data)
elif isinstance(data, list):
return [self.dtons(e) for e in data]
return data
def parser_cmd_init(self):
self.parser.usage = 'sunhpc %s [options]' % self.subcommand
if self.subdesc:
self.parser.description = self.subdesc
# 重新设置 formatter 为 RawTextHelpFormatter
self.parser.formatter_class = argparse.RawTextHelpFormatter
if self.epilog is None:
epilogmsg = "\n"
epilogmsg += "Help & Documentation:\n"
epilogmsg += " - Gitea Repository : https://gitea.sunhpc.com/py/sunhpc\n"
epilogmsg += " - Documentation : https://gitea.sunhpc.com/py/sunhpc/wiki\n"
epilogmsg += " - Issue Tracker : https://gitea.sunhpc.com/py/sunhpc/issues\n"
#epilogmsg += "SunHPC Tool v1.0.0\n"
#epilogmsg += "Copyright (c) 2026 SunHPC Team\n"
#epilogmsg += "This is open-source software released under the Apache 2.0 license.\n"
self.parser.epilog = epilogmsg
def parser_cmd_args(self, args):
original_stderr = sys.stderr
sys.stderr = StringIO()
try:
parsed, unknown = self.parser.parse_known_args(args)
result = vars(parsed) # 返回字典格式
result['unknown'] = unknown
return result
#return vars(parsed) # 返回字典格式
except SystemExit as e:
error_msg = sys.stderr.getvalue()
sys.stderr = original_stderr
if e.code != 0:
self.log.error(error_msg)
#pass
#return {}
#return {}, args
def clearText(self): def clearText(self):
self.text = '' self.text = ''
def addText(self, s): def addText(self, s):
@@ -600,6 +675,9 @@ class Command:
return rlist return rlist
def runWrapper(self, name, args): def runWrapper(self, name, args):
self.subcommand = name
if args: if args:
command = '%s %s' % (name, ' '.join(args)) command = '%s %s' % (name, ' '.join(args))
else: else:

View File

@@ -3,6 +3,8 @@ import re
import sys import sys
import sunhpc.commands import sunhpc.commands
from sunhpc.configs import Config
class Command(sunhpc.commands.init.command): class Command(sunhpc.commands.init.command):
""" """
这个命令是初始化系统的相关配置,并且写入到配置文件. 这个命令是初始化系统的相关配置,并且写入到配置文件.
@@ -22,6 +24,12 @@ class Command(sunhpc.commands.init.command):
提供一个系统ISO文件、或者挂载路径、例如: /mnt/cdrom 提供一个系统ISO文件、或者挂载路径、例如: /mnt/cdrom
</param> </param>
<arguments>
-h --help 显示帮助信息
-d --debug 开启调试模式
-v --verbose 显示详细信息
</arguments>
<example cmd='init config eth1 /mnt/cdrom'> <example cmd='init config eth1 /mnt/cdrom'>
使用eth1网络接口、和 /mnt/cdrom 挂载路径 使用eth1网络接口、和 /mnt/cdrom 挂载路径
</example> </example>
@@ -30,6 +38,80 @@ class Command(sunhpc.commands.init.command):
使用eth1网络接口、和 /mnt/cdrom 挂载路径 使用eth1网络接口、和 /mnt/cdrom 挂载路径
</example> </example>
""" """
def run(self, params, args): def run(self, params, args):
print ('This is init opt ...') self.parser_cmd_init()
self.parser.add_argument('-i', '--iface', action='store', help='Interface name')
self.parser.add_argument('-d', '--debug', action='store_true', help='Debug mode')
self.parser.add_argument('-f', '--file', action='store_true', help='Configuration file')
self.parser.add_argument(
'iface',
help='Network interface (required, e.g., eth0, eth1)'
)
self.parser.add_argument(
'mnt',
nargs='?',
default='/mnt/cdrom',
help='Mount point (optional, default: /mnt/cdrom)'
)
parsed_params = self.parser_cmd_args(args)
self.log.info('File "/opt/sunhpc/lib/sunhpc/sunhpc/commands/init/config/__init__.py", line 63, in run File "/opt/sunhpc/lib/sunhpc/sunhpc/commands/init/config/__init__.py", line 63, in run')
#print (parsed_params)
iface = parsed_params.get('iface', 'eth0')
'''
config = Config()
config.iface = 'eth1'
config.mnt = '/mnt/cdrom'
config.network = '1.1.1.1'
config.netmask = '255.255.255.0'
config.gateway = '1.1.1.1'
config.dhcp = Config()
config.dhcp.enabled = True
config.dhcp.server = '1.1.1.1'
config.dhcp.options = ['domain-name', 'domain-name-servers']
config.http = Config()
config.http.enabled = True
config.http.server = '1.1.1.1'
config.api = Config()
config.api.enabled = True
config.api.server = '1.1.1.1'
config.save('/tmp/sunhpc.yaml')
self.net.setiface('eth0')
print (f'Interface: {self.net.iface}')
print (f'IP : {self.net.getip()}')
print (f'Network: {self.net.getnetwork()}')
print (f'Netmask: {self.net.getnetmask()}')
print (f'Gateway: {self.net.getgateway()}')
print (f'MAC : {self.net.getmac()}')
print (f'CIDR : {self.net.getcidr()}')
print (f'IPv6 : {self.net.getipv6()}')
rx, tx = self.net.get_transfer_human()
print (f'RX : {rx}')
print (f'TX : {tx}')
#print (self.disk.get_disk_info('sda'))
#print (self.mem.get_memory_summary())
#print (self.mem.get_memory_info())
#self.fmt.add_header(['Interface', 'IP', 'Network', 'Netmask', 'Gateway', 'MAC', 'CIDR', 'IPv6'])
d1 = {
'name', 'alice',
'age', 18,
'city', 'beijing',
'occupation', 'student'
}
self.fmt.dict_output(d1)
'''

View File

@@ -0,0 +1,378 @@
import os
import yaml
from typing import Any, Optional, Dict, List
from pathlib import Path
from sunhpc.logs import Logs
slog = Logs(name="Config", show_time=True)
class Config:
"""功能完整的配置管理器(支持动态创建属性)"""
def __init__(self, config_path: Optional[str] = None, auto_load: bool = True):
self._config_path = config_path
self._data = {}
self._initialized = False
if config_path and auto_load and os.path.exists(config_path):
self.load(config_path)
else:
# 初始化空配置
self._initialized = True
def load(self, path: str):
"""从 YAML 文件加载配置"""
with open(path, 'r', encoding='utf-8') as f:
self._data = yaml.safe_load(f)
# 清空现有属性
for attr in list(self.__dict__.keys()):
if not attr.startswith('_'):
delattr(self, attr)
self._build_attributes()
self._initialized = True
def _build_attributes(self):
"""构建属性访问"""
self._build_from_dict(self._data)
def _build_from_dict(self, data: dict, parent_obj=None):
"""递归构建属性"""
if parent_obj is None:
parent_obj = self
for key, value in data.items():
# 处理键名中的空格
clean_key = key.replace(' ', '_')
if isinstance(value, dict):
# 创建子对象
sub_obj = Config()
sub_obj._build_from_dict(value)
setattr(parent_obj, clean_key, sub_obj)
elif isinstance(value, list):
# 转换列表中的字典
converted_list = []
for item in value:
if isinstance(item, dict):
sub_obj = Config()
sub_obj._build_from_dict(item)
converted_list.append(sub_obj)
else:
converted_list.append(item)
setattr(parent_obj, clean_key, converted_list)
else:
setattr(parent_obj, clean_key, value)
def __setattr__(self, name: str, value: Any):
"""拦截属性设置,支持动态创建配置项"""
# 允许设置私有属性和已初始化的内部属性
if name.startswith('_') or name in ['_config_path', '_data', '_initialized']:
super().__setattr__(name, value)
return
# 动态创建配置项
if isinstance(value, dict):
# 如果设置的是字典,转换为 Config 对象
sub_obj = Config()
sub_obj._build_from_dict(value)
super().__setattr__(name, sub_obj)
elif isinstance(value, list):
# 处理列表
converted_list = []
for item in value:
if isinstance(item, dict):
sub_obj = Config()
sub_obj._build_from_dict(item)
converted_list.append(sub_obj)
else:
converted_list.append(item)
super().__setattr__(name, converted_list)
else:
super().__setattr__(name, value)
# 标记数据已修改
self._initialized = True
def __getattr__(self, name: str):
"""访问不存在的属性时返回 None 而不是报错"""
# 如果属性不存在,返回 None可选行为
return None
def save(self, path: Optional[str] = None):
"""保存配置到文件"""
save_path = path or self._config_path
if not save_path:
raise ValueError("未指定保存路径")
# 确保目录存在
Path(save_path).parent.mkdir(parents=True, exist_ok=True)
with open(save_path, 'w', encoding='utf-8') as f:
yaml.dump(self.to_dict(), f, default_flow_style=False, indent=2, allow_unicode=True)
print(f"配置已保存到: {save_path}")
def to_dict(self) -> dict:
"""转换回字典"""
result = {}
for key, value in self.__dict__.items():
# 跳过私有属性
if key.startswith('_'):
continue
if isinstance(value, Config):
result[key] = value.to_dict()
elif isinstance(value, list):
converted_list = []
for item in value:
if isinstance(item, Config):
converted_list.append(item.to_dict())
else:
converted_list.append(item)
result[key] = converted_list
else:
result[key] = value
return result
def create_section(self, section_name: str):
"""创建一个新的配置节"""
if not hasattr(self, section_name):
setattr(self, section_name, Config())
return getattr(self, section_name)
def set_nested(self, path: str, value: Any):
"""通过点号路径设置值,例如: set_nested('network.interface', 'eth0')"""
parts = path.split('.')
obj = self
# 导航到父对象
for part in parts[:-1]:
if not hasattr(obj, part):
setattr(obj, part, Config())
obj = getattr(obj, part)
# 设置最终值
setattr(obj, parts[-1], value)
def get_nested(self, path: str, default: Any = None) -> Any:
"""通过点号路径获取值"""
parts = path.split('.')
obj = self
for part in parts:
if not hasattr(obj, part):
return default
obj = getattr(obj, part)
return obj
def __repr__(self):
return f"Config({self.to_dict()})"
def print_tree(self, indent: int = 0):
"""打印配置树(调试用)"""
for key, value in self.__dict__.items():
if key.startswith('_'):
continue
print(' ' * indent + f"{key}:", end=' ')
if isinstance(value, Config):
print()
value.print_tree(indent + 1)
elif isinstance(value, list):
print(f"[{len(value)} items]")
for i, item in enumerate(value):
print(' ' * (indent + 1) + f"[{i}]:", end=' ')
if isinstance(item, Config):
print()
item.print_tree(indent + 2)
else:
print(item)
else:
print(value)
# ============= 使用示例 =============
def main():
# 1. 创建新配置(不加载现有文件)
config = Config() # 空配置
# 2. 直接设置属性(会自动创建)
config.interface = 'eth1'
config.network = '1.1.1.1'
config.netmask = '255.255.255.0'
config.gateway = '192.168.1.1'
# 3. 创建嵌套配置
config.dhcp = Config()
config.dhcp.enabled = True
config.dhcp.start = '192.168.1.100'
config.dhcp.end = '192.168.1.200'
# 4. 或者通过点号路径设置(更简洁)
config.set_nested('warewulf.port', 9873)
config.set_nested('warewulf.secure', False)
config.set_nested('warewulf.autobuild_overlays', True)
# 5. 创建列表配置
config.dns_servers = ['8.8.8.8', '8.8.4.4']
# 6. 创建复杂的列表字典结构
config.mounts = [
{'source': '/etc/resolv.conf', 'dest': '/etc/resolv.conf', 'readonly': True},
{'source': '/etc/hosts', 'dest': '/etc/hosts', 'readonly': True}
]
# 7. 使用 create_section 方法
nfs_section = config.create_section('nfs')
nfs_section.enabled = True
nfs_section.export_paths = [
{'path': '/home', 'options': 'rw,sync'},
{'path': '/opt', 'options': 'ro,sync'}
]
# 8. 直接访问和修改
print("=" * 50)
print("配置访问示例:")
print(f"interface: {config.interface}")
print(f"network: {config.network}")
print(f"DHCP enabled: {config.dhcp.enabled}")
print(f"Warewulf port: {config.warewulf.port}")
print(f"DNS servers: {config.dns_servers}")
# 9. 修改已存在的配置
config.network = '2.2.2.0'
config.dhcp.enabled = False
config.warewulf.secure = True
# 10. 添加更多配置
config.api = Config()
config.api.enabled = True
config.api.allowed_subnets = ['10.0.0.0/8', '192.168.0.0/16']
config.ssh = Config()
config.ssh.key_types = ['ed25519', 'rsa', 'ecdsa']
# 11. 保存配置
config.save('config_initial.yaml')
# 12. 重新加载配置
print("\n" + "=" * 50)
print("重新加载配置:")
new_config = Config('config_initial.yaml')
print(f"加载的 interface: {new_config.interface}")
print(f"加载的 network: {new_config.network}")
print(f"加载的 DHCP enabled: {new_config.dhcp.enabled}")
print(f"加载的 Warewulf port: {new_config.warewulf.port}")
# 13. 配置树形展示
print("\n" + "=" * 50)
print("配置树形结构:")
new_config.print_tree()
# 14. 使用 get_nested 安全访问
print("\n" + "=" * 50)
print("安全访问示例:")
port = new_config.get_nested('warewulf.port', 9090)
print(f"warewulf.port: {port}")
not_exist = new_config.get_nested('nonexist.key', 'default_value')
print(f"不存在的键: {not_exist}")
# ============= 批量初始化配置 =============
class ConfigInitializer:
"""配置初始化器,支持字典或关键字参数批量设置"""
@staticmethod
def from_dict(config_dict: dict) -> Config:
"""从字典创建配置"""
config = Config()
ConfigInitializer._dict_to_config(config, config_dict)
return config
@staticmethod
def _dict_to_config(config_obj: Config, data: dict):
"""递归将字典转换为配置对象"""
for key, value in data.items():
if isinstance(value, dict):
sub_obj = Config()
ConfigInitializer._dict_to_config(sub_obj, value)
setattr(config_obj, key, sub_obj)
elif isinstance(value, list):
converted_list = []
for item in value:
if isinstance(item, dict):
sub_obj = Config()
ConfigInitializer._dict_to_config(sub_obj, item)
converted_list.append(sub_obj)
else:
converted_list.append(item)
setattr(config_obj, key, converted_list)
else:
setattr(config_obj, key, value)
@staticmethod
def from_yaml_template(template_path: str) -> Config:
"""从 YAML 模板文件初始化配置"""
with open(template_path, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
return ConfigInitializer.from_dict(data)
# ============= 快速初始化示例 =============
def quick_init_example():
"""快速初始化示例"""
# 方式1: 逐个设置(你要求的方式)
config = Config()
config.interface = 'eth1'
config.network = '1.1.1.1'
config.netmask = '255.255.255.0'
config.gateway = '192.168.1.1'
config.dhcp_enabled = True
config.dhcp_start = '192.168.1.100'
config.dhcp_end = '192.168.1.200'
config.save('config_by_assign.yaml')
print("方式1: 逐个赋值保存完成")
# 方式2: 使用 set_nested 批量设置
config2 = Config()
settings = {
'interface': 'eth0',
'network': '10.0.0.0',
'netmask': '255.0.0.0',
'dhcp.enabled': True,
'dhcp.range_start': '10.0.0.100',
'dhcp.range_end': '10.0.0.200',
'warewulf.port': 9873,
'warewulf.secure': True
}
for path, value in settings.items():
config2.set_nested(path, value)
config2.save('config_by_set_nested.yaml')
print("方式2: set_nested 批量设置完成")
# 方式3: 从字典初始化
config_dict = {
'interface': 'eth2',
'network': '172.16.0.0',
'netmask': '255.255.240.0',
'dhcp': {
'enabled': True,
'start': '172.16.0.100',
'end': '172.16.0.200'
},
'warewulf': {
'port': 9873,
'secure': False
}
}
config3 = ConfigInitializer.from_dict(config_dict)
config3.save('config_from_dict.yaml')
print("方式3: 从字典初始化完成")
if __name__ == '__main__':
main()
print("\n" + "=" * 50)
quick_init_example()

717
lib/sunhpc/sunhpc/disks.py Normal file
View File

@@ -0,0 +1,717 @@
#!/usr/bin/env python3
"""
DiskInfo class to retrieve comprehensive disk information.
Supports physical disks, SSDs, and virtual disks (VMware, VirtualBox, KVM).
No third-party libraries required, uses /sys, /proc, and system commands.
"""
import os
import re
import glob
import subprocess
from typing import Optional, Dict, List, Tuple, Any
class DiskInfo:
"""Class to retrieve detailed information about disks."""
def __init__(self):
self.disk_name: Optional[str] = None
self._info_cache: Dict[str, Any] = {}
def get_all_disks(self) -> List[str]:
"""
Get list of all disk devices (excluding partitions).
Returns disk names like ['sda', 'sdb', 'vda', 'nvme0n1'].
"""
disks = []
# Method 1: /sys/block/
try:
for path in glob.glob('/sys/block/*'):
disk = os.path.basename(path)
# Skip loop devices, ram disks, and removable drives if needed
if not disk.startswith(('loop', 'ram', 'sr')):
# Check if it's a real disk (has device directory)
if os.path.exists(f'/sys/block/{disk}/device'):
disks.append(disk)
except Exception:
pass
# Method 2: /proc/partitions (for additional disks)
try:
with open('/proc/partitions', 'r') as f:
for line in f:
parts = line.strip().split()
if len(parts) >= 4:
device = parts[3]
# Major number 8 = SCSI/SATA, 9 = MD, 11 = SCSI CD-ROM, 259 = NVMe
major = int(parts[1]) if parts[1].isdigit() else 0
if (major in (8, 259) or device.startswith(('sd', 'hd', 'vd', 'nvme'))) and \
device not in disks and not device[-1].isdigit():
disks.append(device)
except Exception:
pass
return sorted(set(disks))
def set_disk(self, disk_name: str) -> 'DiskInfo':
"""Set specific disk to query."""
self.disk_name = disk_name
self._info_cache = {}
return self
def get_disk_info(self, disk_name: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""
Get comprehensive information for a disk.
If disk_name is None, uses the currently set disk.
Returns dictionary with all available information.
"""
target_disk = disk_name or self.disk_name
if not target_disk:
# Return info for all disks
all_info = {}
for disk in self.get_all_disks():
all_info[disk] = self.get_disk_info(disk)
return all_info
# Check cache
if target_disk in self._info_cache:
return self._info_cache[target_disk]
# Gather all information
info = {
'name': target_disk,
'model': '',
'serial': '',
'size_bytes': None,
'size_human': None,
'firmware': None,
'interface': None,
'type': None, # HDD, SSD, NVMe, Virtual
'rotational': None, # True for HDD, False for SSD
'rpm': None, # Rotational speed for HDD
'sector_size': None,
'sector_size_logical': None,
'sector_size_physical': None,
'smart_supported': None,
'smart_enabled': None,
'temperature': None,
'power_on_hours': None,
'power_on_count': None,
'bad_sectors': None, # Reallocated sectors count
'pending_sectors': None, # Pending reallocation
'uncorrectable_sectors': None,
'health_status': None, # PASSED, FAILED, UNKNOWN
'vendor': '',
'removable': None,
'wwn': '', # World Wide Name
'device_path': None,
'partitions': None,
'is_virtual': False,
'virtual_type': None, # VMware, VirtualBox, KVM, etc.
}
# Try to get information from sysfs (most reliable for basic info)
self._get_info_from_sysfs(target_disk, info)
# Try to get SMART information (requires smartctl)
self._get_smart_info(target_disk, info)
# Try to get additional info from /proc/scsi/scsi
self._get_scsi_info(target_disk, info)
# Detect virtualization
self._detect_virtualization(target_disk, info)
# Get partition information
info['partitions'] = self._get_partitions(target_disk)
# Cache the results
self._info_cache[target_disk] = info
return info
def _get_info_from_sysfs(self, disk: str, info: Dict[str, Any]):
"""Get disk information from sysfs (/sys/block/)."""
base_path = f'/sys/block/{disk}'
if not os.path.exists(base_path):
return
# Device path
info['device_path'] = f'/dev/{disk}'
# Size in sectors (each sector usually 512 bytes)
size_path = f'{base_path}/size'
if os.path.exists(size_path):
try:
with open(size_path, 'r') as f:
sectors = int(f.read().strip())
sector_size = 512 # Default standard
# Try to get actual sector size
if os.path.exists(f'{base_path}/queue/hw_sector_size'):
with open(f'{base_path}/queue/hw_sector_size', 'r') as f2:
sector_size = int(f2.read().strip())
info['size_bytes'] = sectors * sector_size
info['size_human'] = self._bytes_to_human(info['size_bytes'])
info['sector_size'] = sector_size
except Exception:
pass
# Logical and physical sector size
logical_path = f'{base_path}/queue/logical_block_size'
physical_path = f'{base_path}/queue/physical_block_size'
if os.path.exists(logical_path):
try:
with open(logical_path, 'r') as f:
info['sector_size_logical'] = int(f.read().strip())
except Exception:
pass
if os.path.exists(physical_path):
try:
with open(physical_path, 'r') as f:
info['sector_size_physical'] = int(f.read().strip())
except Exception:
pass
# Rotational (spinning disk vs SSD)
rotational_path = f'{base_path}/queue/rotational'
if os.path.exists(rotational_path):
try:
with open(rotational_path, 'r') as f:
rotational = int(f.read().strip())
info['rotational'] = rotational == 1
info['type'] = 'HDD' if rotational else 'SSD'
except Exception:
pass
# Removable
removable_path = f'{base_path}/removable'
if os.path.exists(removable_path):
try:
with open(removable_path, 'r') as f:
info['removable'] = int(f.read().strip()) == 1
except Exception:
pass
# Model from device directory
device_path = f'{base_path}/device'
if os.path.exists(device_path):
# Try different model file locations
for model_file in ['model', 'product', 'inquiry', 'device/model']:
full_path = os.path.join(device_path, model_file)
if os.path.exists(full_path):
try:
with open(full_path, 'r') as f:
model = f.read().strip()
if model and not model.isspace():
info['model'] = model
break
except Exception:
pass
# Firmware version
for fw_file in ['fwrev', 'firmware_rev', 'rev']:
full_path = os.path.join(device_path, fw_file)
if os.path.exists(full_path):
try:
with open(full_path, 'r') as f:
info['firmware'] = f.read().strip()
break
except Exception:
pass
# Serial number
for serial_file in ['serial', 'wwid']:
full_path = os.path.join(device_path, serial_file)
if os.path.exists(full_path):
try:
serial = f.read().strip() if os.path.isfile(full_path) else None
if serial and not serial.isspace():
info['serial'] = serial
break
except Exception:
pass
# WWN (World Wide Name)
wwn_path = f'{base_path}/wwid'
if os.path.exists(wwn_path):
try:
with open(wwn_path, 'r') as f:
wwn = f.read().strip()
if wwn.startswith('naa.'):
info['wwn'] = wwn
except Exception:
pass
# Interface type
if os.path.exists(f'{base_path}/device/subsystem'):
try:
subsystem = os.path.realpath(f'{base_path}/device/subsystem')
if 'nvme' in subsystem:
info['interface'] = 'NVMe'
elif 'ata' in subsystem or 'scsi' in subsystem:
# Determine SATA or SAS
if os.path.exists(f'{device_path}/sas_address'):
info['interface'] = 'SAS'
else:
info['interface'] = 'SATA'
elif 'virtio' in subsystem:
info['interface'] = 'VirtIO'
except Exception:
pass
def _get_scsi_info(self, disk: str, info: Dict[str, Any]):
"""Get SCSI disk information from /proc/scsi/scsi."""
try:
with open('/proc/scsi/scsi', 'r') as f:
content = f.read()
# Find matching disk entry
patterns = [
rf'Host:\s+scsi\d+\s+Channel:\s+\d+\s+Id:\s+\d+\s+Lun:\s+\d+\s*\n\s*Vendor:\s+(\S+)\s+Model:\s+(.+?)\s+Rev:\s+(\S+)\s*\n\s*Type:\s+\S+\s+ANSI\s+SCSI\s+revision:\s+\d+',
]
for pattern in patterns:
matches = re.findall(pattern, content, re.MULTILINE)
for vendor, model, rev in matches:
if disk in model.lower() or disk in vendor.lower():
info['vendor'] = vendor.strip()
if not info['model']:
info['model'] = model.strip()
if not info['firmware']:
info['firmware'] = rev.strip()
break
except Exception:
pass
def _get_smart_info(self, disk: str, info: Dict[str, Any]):
"""
Get SMART information using smartctl command.
This requires smartmontools package to be installed.
"""
if not self._check_command_exists('smartctl'):
info['smart_supported'] = False
return
try:
# Check if SMART is supported
result = subprocess.run(
['smartctl', '-i', f'/dev/{disk}'],
capture_output=True,
text=True,
timeout=10
)
if 'SMART support is:' in result.stdout:
smart_lines = result.stdout
if 'SMART support is: Available' in smart_lines:
info['smart_supported'] = True
if 'SMART support is: Enabled' in smart_lines:
info['smart_enabled'] = True
# Get health information
result = subprocess.run(
['smartctl', '-H', f'/dev/{disk}'],
capture_output=True,
text=True,
timeout=10
)
if 'SMART overall-health self-assessment test result:' in result.stdout:
health_line = re.search(
r'SMART overall-health self-assessment test result:\s+(.+)',
result.stdout
)
if health_line:
info['health_status'] = health_line.group(1).strip()
# Get detailed SMART attributes
result = subprocess.run(
['smartctl', '-A', f'/dev/{disk}'],
capture_output=True,
text=True,
timeout=10
)
# Parse SMART attributes
attributes = {
'Reallocated_Sector_Ct': 'bad_sectors',
'Current_Pending_Sector': 'pending_sectors',
'Offline_Uncorrectable': 'uncorrectable_sectors',
'Temperature_Celsius': 'temperature',
'Power_On_Hours': 'power_on_hours',
'Power_Cycle_Count': 'power_on_count',
'Rotational_Rate': 'rpm'
}
for line in result.stdout.split('\n'):
for attr, key in attributes.items():
if attr in line:
# Parse the attribute value (usually the 10th column)
parts = line.split()
if len(parts) >= 10:
try:
value = parts[9] # RAW_VALUE column
if value.isdigit():
if key == 'rpm' and int(value) > 0:
info['rpm'] = int(value)
if info['type'] == 'HDD':
info['type'] = 'HDD'
elif key != 'rpm':
info[key] = int(value)
except (ValueError, IndexError):
pass
break
# Try to get rotational rate for HDDs from other sources
if info.get('rotational') and not info.get('rpm'):
# Try to get from sysfs
rpm_path = f'/sys/block/{disk}/device/rotational_rate'
if os.path.exists(rpm_path):
try:
with open(rpm_path, 'r') as f:
rpm = int(f.read().strip())
if rpm > 0:
info['rpm'] = rpm
except Exception:
pass
except subprocess.TimeoutExpired:
pass
except FileNotFoundError:
info['smart_supported'] = False
except Exception:
pass
def _detect_virtualization(self, disk: str, info: Dict[str, Any]):
"""Detect if disk is virtual (VMware, VirtualBox, KVM, etc.)."""
info['is_virtual'] = False
# Check disk model for virtualization indicators
model = info.get('model', '').lower()
vendor = info.get('vendor', '').lower()
if 'vmware' in model or 'vmware' in vendor:
info['is_virtual'] = True
info['virtual_type'] = 'VMware'
info['type'] = 'Virtual'
elif 'vbox' in model or 'virtualbox' in model:
info['is_virtual'] = True
info['virtual_type'] = 'VirtualBox'
info['type'] = 'Virtual'
elif 'qemu' in model or 'kvm' in model:
info['is_virtual'] = True
info['virtual_type'] = 'KVM/QEMU'
info['type'] = 'Virtual'
elif 'virtio' in model:
info['is_virtual'] = True
info['virtual_type'] = 'VirtIO'
info['type'] = 'Virtual'
# Check system virtualization
try:
with open('/proc/cpuinfo', 'r') as f:
if 'hypervisor' in f.read().lower():
if not info['is_virtual']:
info['is_virtual'] = True
info['virtual_type'] = 'Unknown Hypervisor'
except Exception:
pass
def _get_partitions(self, disk: str) -> Optional[List[Dict[str, Any]]]:
"""Get partition information for the disk."""
partitions = []
try:
# Read from /proc/partitions
with open('/proc/partitions', 'r') as f:
for line in f:
parts = line.strip().split()
if len(parts) >= 4:
device = parts[3]
if device.startswith(disk) and device != disk:
size_kb = int(parts[2]) if parts[2].isdigit() else 0
partitions.append({
'name': device,
'size_kb': size_kb,
'size_human': self._bytes_to_human(size_kb * 1024)
})
except Exception:
pass
return partitions if partitions else None
def get_disk_summary(self, disk_name: Optional[str] = None) -> str:
"""Get a human-readable summary for a disk."""
info = self.get_disk_info(disk_name)
if not info:
return f"No information available for disk: {disk_name}"
if isinstance(info, dict) and 'name' in info:
# Single disk
return self._format_disk_summary(info)
elif isinstance(info, dict):
# Multiple disks
summaries = []
for disk, disk_info in info.items():
summaries.append(self._format_disk_summary(disk_info))
return '\n\n'.join(summaries)
return "No disk information available"
def _format_disk_summary(self, info: Dict[str, Any]) -> str:
"""Format disk information as a human-readable string."""
lines = []
lines.append(f"Disk: {info['name']} ({info['device_path']})")
lines.append(f"{'=' * 60}")
if info.get('model'):
lines.append(f" Model: {info['model']}")
if info.get('vendor'):
lines.append(f" Vendor: {info['vendor']}")
if info.get('serial'):
lines.append(f" Serial Number: {info['serial']}")
if info.get('firmware'):
lines.append(f" Firmware: {info['firmware']}")
if info.get('wwn'):
lines.append(f" WWN: {info['wwn']}")
lines.append(f" Type: {info.get('type', 'Unknown')}")
if info.get('rotational') is not None:
lines.append(f" Rotational: {'Yes' if info['rotational'] else 'No'}")
if info.get('rpm'):
lines.append(f" RPM: {info['rpm']}")
if info.get('size_human'):
lines.append(f" Size: {info['size_human']} ({info.get('size_bytes', 0):,} bytes)")
if info.get('interface'):
lines.append(f" Interface: {info['interface']}")
if info.get('sector_size'):
lines.append(f" Sector Size: {info['sector_size']} bytes")
if info.get('sector_size_logical'):
lines.append(f" Logical Sector Size: {info['sector_size_logical']} bytes")
if info.get('sector_size_physical'):
lines.append(f" Physical Sector Size: {info['sector_size_physical']} bytes")
lines.append(f" Removable: {'Yes' if info.get('removable') else 'No'}")
lines.append(f" Virtual: {'Yes' if info.get('is_virtual') else 'No'}")
if info.get('virtual_type'):
lines.append(f" Virtual Type: {info['virtual_type']}")
# SMART information
if info.get('smart_supported'):
lines.append(" SMART:")
lines.append(f" Supported: Yes")
lines.append(f" Enabled: {'Yes' if info.get('smart_enabled') else 'No'}")
if info.get('health_status'):
lines.append(f" Health: {info['health_status']}")
if info.get('temperature'):
lines.append(f" Temperature: {info['temperature']}°C")
if info.get('power_on_hours'):
days = info['power_on_hours'] // 24
lines.append(f" Power On Hours: {info['power_on_hours']} hours ({days} days)")
if info.get('power_on_count'):
lines.append(f" Power Cycle Count: {info['power_on_count']}")
# Bad sector information
bad_info = []
if info.get('bad_sectors'):
bad_info.append(f"Reallocated: {info['bad_sectors']}")
if info.get('pending_sectors'):
bad_info.append(f"Pending: {info['pending_sectors']}")
if info.get('uncorrectable_sectors'):
bad_info.append(f"Uncorrectable: {info['uncorrectable_sectors']}")
if bad_info:
lines.append(f" Bad Sectors: {', '.join(bad_info)}")
else:
lines.append(" SMART: Not supported or smartctl not installed")
# Partitions
if info.get('partitions'):
lines.append(f" Partitions: {len(info['partitions'])}")
for part in info['partitions'][:5]: # Show first 5
lines.append(f" - {part['name']}: {part['size_human']}")
if len(info['partitions']) > 5:
lines.append(f" ... and {len(info['partitions']) - 5} more")
return '\n'.join(lines)
def check_disk_health(self, disk_name: Optional[str] = None) -> Dict[str, Any]:
"""
Check disk health and return a summary.
Returns dictionary with health status and recommendations.
"""
info = self.get_disk_info(disk_name)
if not info or (isinstance(info, dict) and 'name' not in info):
return {'error': 'No disk information available'}
if isinstance(info, dict) and 'name' in info:
disks_info = {info['name']: info}
else:
disks_info = info
health_summary = {}
for disk_name, disk_info in disks_info.items():
health = {
'disk': disk_name,
'status': 'UNKNOWN',
'issues': [],
'warnings': [],
'recommendations': []
}
# Check SMART health status
if disk_info.get('health_status'):
if 'PASSED' in disk_info['health_status']:
health['status'] = 'GOOD'
elif 'FAILED' in disk_info['health_status']:
health['status'] = 'FAILED'
health['issues'].append('SMART health test failed')
health['recommendations'].append('Backup data immediately and replace disk')
# Check bad sectors
bad_sectors = disk_info.get('bad_sectors', 0)
pending = disk_info.get('pending_sectors', 0)
uncorrectable = disk_info.get('uncorrectable_sectors', 0)
if bad_sectors > 0:
health['warnings'].append(f'Reallocated sectors: {bad_sectors}')
if bad_sectors > 100:
health['issues'].append(f'High number of reallocated sectors: {bad_sectors}')
health['recommendations'].append('Consider replacing disk soon')
if pending > 0:
health['warnings'].append(f'Pending sectors: {pending}')
health['recommendations'].append('Run full disk surface test')
if uncorrectable > 0:
health['issues'].append(f'Uncorrectable sectors: {uncorrectable}')
health['recommendations'].append('Disk may have physical damage, replace immediately')
# Check temperature
temp = disk_info.get('temperature')
if temp:
if temp > 70:
health['warnings'].append(f'High temperature: {temp}°C')
health['recommendations'].append('Improve cooling')
elif temp > 60:
health['warnings'].append(f'Elevated temperature: {temp}°C')
# Check power on hours (for HDDs)
if disk_info.get('type') == 'HDD' and disk_info.get('power_on_hours'):
hours = disk_info['power_on_hours']
years = hours / 8760 # 365 * 24
if years > 5:
health['warnings'].append(f'Disk age: {years:.1f} years of power-on time')
health['recommendations'].append('Consider replacing due to age')
if not health['issues'] and not health['warnings']:
health['status'] = 'EXCELLENT'
elif not health['issues']:
health['status'] = 'WARNING'
health_summary[disk_name] = health
return health_summary
@staticmethod
def _check_command_exists(command: str) -> bool:
"""Check if a command exists in the system."""
try:
subprocess.run(
['which', command],
capture_output=True,
text=True,
timeout=5
)
return True
except Exception:
return False
@staticmethod
def _bytes_to_human(bytes_val: Optional[int]) -> str:
"""Convert bytes to human-readable format."""
if bytes_val is None:
return "Unknown"
for unit in ['B', 'KB', 'MB', 'GB', 'TB', 'PB']:
if bytes_val < 1024.0:
return f"{bytes_val:.2f} {unit}"
bytes_val /= 1024.0
return f"{bytes_val:.2f} EB"
def refresh(self) -> 'DiskInfo':
"""Clear cache and refresh disk information."""
self._info_cache = {}
return self
def __repr__(self) -> str:
if self.disk_name:
return f"DiskInfo(disk={self.disk_name})"
else:
disks = self.get_all_disks()
return f"DiskInfo(disks={disks})"
# ------------------------------------------------------------------
# Example usage and test
# ------------------------------------------------------------------
if __name__ == '__main__':
disk = DiskInfo()
print("=== All Disks ===")
all_disks = disk.get_all_disks()
print(f"Found disks: {', '.join(all_disks)}")
print()
# Get summary for all disks
print(disk.get_disk_summary())
print()
# Get detailed information for each disk
for disk_name in all_disks:
print(f"\n{'=' * 80}")
print(f"Detailed information for {disk_name}:")
print(f"{'=' * 80}")
info = disk.get_disk_info(disk_name)
if info:
print(f"\nJSON-like structure (partial):")
for key, value in info.items():
if key != 'partitions' and value is not None:
print(f" {key}: {value}")
if info.get('partitions'):
print(f" partitions: {len(info['partitions'])} partitions found")
# Check health for all disks
print(f"\n{'=' * 80}")
print("Disk Health Check")
print(f"{'=' * 80}")
health = disk.check_disk_health()
for disk_name, status in health.items():
print(f"\n{disk_name}:")
print(f" Status: {status['status']}")
if status['warnings']:
print(f" Warnings: {', '.join(status['warnings'])}")
if status['issues']:
print(f" Issues: {', '.join(status['issues'])}")
if status['recommendations']:
print(f" Recommendations: {', '.join(status['recommendations'])}")
# Example of setting a specific disk
if all_disks:
print(f"\n{'=' * 80}")
print(f"Working with specific disk: {all_disks[0]}")
print(f"{'=' * 80}")
disk.set_disk(all_disks[0])
print(f"Disk model: {disk.get_disk_info().get('model', 'Unknown')}")
print(f"Disk size: {disk.get_disk_info().get('size_human', 'Unknown')}")

View File

@@ -0,0 +1,84 @@
from enum import Enum
from typing import Dict, Any, Union, Optional
class ErrorCode(Enum):
'''错误码枚举'''
# status code
SUCCESS = '000'
FAIL = '001'
# command error
INVALID_COMMAND = '010'
INVALID_FORMAT = '012'
# config error
EMPTY_CONFIG = '020'
# File error
FILE_NOT_FOUND_ERROR = '030'
# module error
IMPORT_ERROR = '040'
# network error
NETWORK_ERROR = '050'
# timeout error
TIMEOUT_ERROR = '100'
# yaml error
YAML_PARSE_ERROR = '402'
@classmethod
def get_all_codes(cls):
return [code.value for code in cls]
class ErrorMessages:
'''错误消息枚举'''
_MESSAGES = {
ErrorCode.SUCCESS: 'success',
ErrorCode.FAIL: 'fail',
ErrorCode.INVALID_COMMAND: 'invalid command',
ErrorCode.EMPTY_CONFIG: 'empty config',
ErrorCode.INVALID_FORMAT: 'invalid format',
ErrorCode.FILE_NOT_FOUND_ERROR: 'file not found',
ErrorCode.IMPORT_ERROR: 'import error',
ErrorCode.NETWORK_ERROR: 'network error',
ErrorCode.TIMEOUT_ERROR: 'timeout error',
ErrorCode.YAML_PARSE_ERROR: 'yaml parse error',
}
@classmethod
def get_message(cls, err_code: Union[ErrorCode, str]) -> str:
if isinstance(err_code, str):
try:
error_code = ErrorCode(err_code)
except ValueError:
return f'unknown error code: {err_code}'
return cls._MESSAGES.get(error_code, f'unknown error code: {err_code}')
SUCCESS = ErrorCode.SUCCESS.value
FAIL = ErrorCode.FAIL.value
INVALID_COMMAND = ErrorCode.INVALID_COMMAND.value
INVALID_FORMAT = ErrorCode.INVALID_FORMAT.value
EMPTY_CONFIG = ErrorCode.EMPTY_CONFIG.value
FILE_NOT_FOUND_ERROR = ErrorCode.FILE_NOT_FOUND_ERROR.value
IMPORT_ERROR = ErrorCode.IMPORT_ERROR.value
NETWORK_ERROR = ErrorCode.NETWORK_ERROR.value
TIMEOUT_ERROR = ErrorCode.TIMEOUT_ERROR.value
YAML_PARSE_ERROR = ErrorCode.YAML_PARSE_ERROR.value
def is_error(data: Any) -> bool:
'''
判断数据是否为错误码
: param data: 待判断数据
: return: True表示错误码False表示正常码
'''
if not isinstance(data, str):
return False
if len(data) == 3 and data.isdigit():
return data in [ c.value for c in ErrorCode if c != ErrorCode.SUCCESS]
return False

528
lib/sunhpc/sunhpc/logs.py Normal file
View File

@@ -0,0 +1,528 @@
from re import I
import sys
import time
from datetime import datetime
from typing import Optional, Any, Dict
from sunhpc.util import getTerminalWidth
class Colors:
"""终端颜色代码"""
# 基本颜色
BLACK = '\033[30m'
RED = '\033[31m'
GREEN = '\033[32m'
YELLOW = '\033[33m'
BLUE = '\033[34m'
MAGENTA = '\033[35m'
CYAN = '\033[36m'
WHITE = '\033[37m'
# 亮色
BRIGHT_BLACK = '\033[90m'
BRIGHT_RED = '\033[91m'
BRIGHT_GREEN = '\033[92m'
BRIGHT_YELLOW = '\033[93m'
BRIGHT_BLUE = '\033[94m'
BRIGHT_MAGENTA = '\033[95m'
BRIGHT_CYAN = '\033[96m'
BRIGHT_WHITE = '\033[97m'
# 背景色
BG_RED = '\033[41m'
BG_GREEN = '\033[42m'
BG_YELLOW = '\033[43m'
BG_BLUE = '\033[44m'
BG_MAGENTA = '\033[45m'
BG_CYAN = '\033[46m'
BG_WHITE = '\033[47m'
# 样式
BOLD = '\033[1m'
DIM = '\033[2m'
ITALIC = '\033[3m'
UNDERLINE = '\033[4m'
BLINK = '\033[5m'
REVERSE = '\033[7m'
HIDDEN = '\033[8m'
# 重置
RESET = '\033[0m'
class LogLevel:
"""日志级别定义"""
DEBUG = 0
INFO = 1
OK = 2
WARN = 3
ERROR = 4
CRITICAL = 5
_names = {
DEBUG: "DEBUG",
INFO: "INFO",
OK: "OK",
WARN: "WARN",
ERROR: "ERROR",
CRITICAL: "CRITICAL"
}
_prefixes = {
DEBUG: "[d]",
INFO: "[i]",
OK: "[+]",
WARN: "[-]",
ERROR: "[e]",
CRITICAL: "[!]"
}
_colors = {
DEBUG: Colors.BRIGHT_BLACK,
INFO: Colors.BRIGHT_BLUE,
OK: Colors.BRIGHT_GREEN,
WARN: Colors.BRIGHT_YELLOW,
ERROR: Colors.BRIGHT_RED,
CRITICAL: Colors.BG_RED + Colors.BRIGHT_WHITE + Colors.BOLD
}
@classmethod
def get_name(cls, level: int) -> str:
return cls._names.get(level, "UNKNOWN")
@classmethod
def get_prefix(cls, level: int) -> str:
return cls._prefixes.get(level, "[?]")
@classmethod
def get_color(cls, level: int) -> str:
return cls._colors.get(level, Colors.WHITE)
def __str__(self):
return self.get_name(self.level)
class Logs:
"""日志类"""
def __init__(self,
name: str = "",
level: int = LogLevel.INFO,
show_time: bool = True,
show_level: bool = False,
show_caller: bool = False,
show_function: bool = False,
time_format: str = "%H:%M:%S",
output_file: Optional[str] = None,
color_enabled: bool = True):
"""
初始化日志对象
:param name: 日志名称(通常为模块名)
:param level: 最低日志级别
:param show_time: 是否显示时间
:param show_level: 是否显示日志级别名称
:param time_format: 时间格式
:param color_enabled: 是否启用颜色
"""
self.name = name
self.level = level
self.reduce = 20
self.show_time = show_time
self.show_level = show_level
self.output_file = output_file
self.show_caller = show_caller
self.show_function = show_function
self.time_format = time_format
self.color_enabled = color_enabled and self._supports_color()
self._file_handler = None
if self.output_file:
try:
self._file_handler = open(self.output_file, 'a', encoding='utf-8')
except Exception:
print(f"Failed to open log file {self.output_file}: {e}", file=sys.stderr)
def __del__(self):
if self._file_handler:
self._file_handler.close()
def _get_caller_info(self, stack_level: int = 3) -> Dict[str, str]:
"""
Get caller information from the stack.
Args:
stack_level: How many levels to go up the stack (2 = direct caller)
Returns:
Dictionary with file, line, function information
"""
try:
# Get the current stack frame
frame = sys._getframe(stack_level)
# Get caller information
filename = frame.f_code.co_filename
line_number = frame.f_lineno
function_name = frame.f_code.co_name
# Get relative path (remove current working directory)
try:
cwd = os.getcwd()
if filename.startswith(cwd):
filename = os.path.relpath(filename, cwd)
except Exception:
pass
return {
'file': filename,
'line': str(line_number),
'function': function_name,
'full_path': filename
}
except Exception:
return {
'file': 'unknown',
'line': '?',
'function': 'unknown',
'full_path': 'unknown'
}
def _supports_color(self) -> bool:
"""检测终端是否支持颜色"""
return (
hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()
) or 'ANSICON' in sys.environ or 'PYCHARM_HOSTED' in sys.environ
def _format_message(self,
msg: str,
level: int,
same_color: Optional[str] = None,
caller_info: Optional[Dict[str, str]] = None) -> str:
"""
格式化日志消息
:param msg: 日志消息
:param level: 日志级别
:param same_color: 统一颜色(如果指定,则忽略级别颜色)
:param caller_info: 调用者信息(文件、行号、函数名、完整路径,可选)
:return: 格式化后的消息
"""
parts = []
# 时间戳
if self.show_time:
timestamp = datetime.now().strftime(self.time_format)
if self.color_enabled:
parts.append(f"{Colors.BRIGHT_BLACK}{timestamp}{Colors.RESET}")
else:
parts.append(timestamp)
# 日志名称
if self.name:
if self.color_enabled:
parts.append(f"{Colors.BRIGHT_CYAN}[{self.name}]{Colors.RESET}")
else:
parts.append(f"[{self.name}]")
# 日志级别名称
if self.show_level:
level_name = LogLevel.get_name(level)
if self.color_enabled:
color = same_color if same_color else LogLevel.get_color(level)
parts.append(f"{color}{level_name:7}{Colors.RESET}")
else:
parts.append(f"{level_name:7}")
# 调用者信息
if self.show_caller and caller_info:
parts.append(f"{caller_info['file']}:{caller_info['line']}")
# 日志级别前缀
prefix = LogLevel.get_prefix(level)
if self.color_enabled:
color = same_color if same_color else LogLevel.get_color(level)
prefix_colored = f"{color}{prefix}{Colors.RESET}"
parts.append(prefix_colored)
else:
parts.append(prefix)
# 消息内容
if self.color_enabled and same_color:
parts.append(f"{same_color}{msg}{Colors.RESET}")
elif self.color_enabled:
color = LogLevel.get_color(level)
parts.append(f"{color}{msg}{Colors.RESET}")
else:
parts.append(msg)
return " ".join(parts)
def _split_words(self, msg: str) -> list:
"""将消息按宽度拆分成多个行"""
try:
terminal_width = (getTerminalWidth() - 30)
except:
terminal_width = 80
if type(msg) in [type([]), type(())]:
msg = ' '.join(msg)
msg = ' '.join(msg.strip('\n').splitlines())
if len(msg) <= terminal_width:
return [msg]
words = msg.split()
result = []
current_line = ''
for word in words:
if current_line:
test_line = current_line + " " + word
else:
test_line = word
if len(test_line) <= terminal_width:
current_line = test_line
else:
if current_line:
result.append(current_line)
if len(word) > terminal_width:
for i in range(0, len(word), terminal_width):
result.append(word[i:i+terminal_width])
current_line = ''
else:
current_line = word
if current_line:
result.append(current_line)
return result
def _log(self, level: int, msg: str, same_color: Optional[str] = None):
"""内部日志方法"""
if level < self.level:
return
# Get caller information (skip _log and the calling method)
caller_info = self._get_caller_info()
intput_lines = self._split_words(msg)
output_lines = []
for line in intput_lines:
output_lines.append(self._format_message(line, level, same_color, caller_info))
# 写入文件
if self._file_handler:
try:
self._file_handler.write(formatted_msg + '\n')
self._file_handler.flush()
except Exception:
pass
# 错误级别输出到 stderr
if level >= LogLevel.ERROR:
for line in output_lines:
if line.strip():
print (line, file=sys.stderr)
#print(formatted_msg, file=sys.stderr)
else:
for line in output_lines:
if line.strip():
print (line)
def set_level(self, level: int):
"""设置日志级别"""
self.level = level
def set_output_file(self, filename: str) -> None:
'''Set or change the output file.'''
if self._file_handler:
self._file_handler.close()
try:
self._file_handler = open(filename, 'a', encoding='utf-8')
self.output_file = filename
except Exception:
pass
def debug(self, msg: str, same_color: Optional[str] = None):
"""调试日志 [d]"""
self._log(LogLevel.DEBUG, msg, same_color)
def info(self, msg: str, same_color: Optional[str] = None):
"""信息日志 [i]"""
self._log(LogLevel.INFO, msg, same_color)
def ok(self, msg: str, same_color: Optional[str] = None):
"""成功日志 [+]"""
self._log(LogLevel.OK, msg, same_color)
def warn(self, msg: str, same_color: Optional[str] = None):
"""警告日志 [-]"""
self._log(LogLevel.WARN, msg, same_color)
def error(self, msg: str, same_color: Optional[str] = None):
"""错误日志 [e]"""
self._log(LogLevel.ERROR, msg, same_color)
def critical(self, msg: str, same_color: Optional[str] = None):
"""严重错误日志 [!]"""
self._log(LogLevel.CRITICAL, msg, same_color)
def __call__(self, msg: str, level: str = "info", same_color: Optional[str] = None):
"""
直接调用方法,支持字符串级别
:param msg: 日志消息
:param level: 日志级别 (debug, info, ok, warn, error, critical)
:param same_color: 统一颜色
"""
level_map = {
"debug": LogLevel.DEBUG,
"info": LogLevel.INFO,
"ok": LogLevel.OK,
"warn": LogLevel.WARN,
"warning": LogLevel.WARN,
"error": LogLevel.ERROR,
"critical": LogLevel.CRITICAL,
"err": LogLevel.ERROR,
}
level_value = level_map.get(level.lower(), LogLevel.INFO)
# 转换颜色字符串到颜色代码(如果提供)
color_code = None
if same_color and self.color_enabled:
color_code = self._get_color_code(same_color)
self._log(level_value, msg, color_code)
def _get_color_code(self, color_name: str) -> str:
"""根据颜色名称获取颜色代码"""
color_map = {
"black": Colors.BLACK,
"red": Colors.RED,
"green": Colors.GREEN,
"yellow": Colors.YELLOW,
"blue": Colors.BLUE,
"magenta": Colors.MAGENTA,
"cyan": Colors.CYAN,
"white": Colors.WHITE,
"bright_black": Colors.BRIGHT_BLACK,
"bright_red": Colors.BRIGHT_RED,
"bright_green": Colors.BRIGHT_GREEN,
"bright_yellow": Colors.BRIGHT_YELLOW,
"bright_blue": Colors.BRIGHT_BLUE,
"bright_magenta": Colors.BRIGHT_MAGENTA,
"bright_cyan": Colors.BRIGHT_CYAN,
"bright_white": Colors.BRIGHT_WHITE,
}
return color_map.get(color_name.lower(), Colors.WHITE)
def line(self, char: str = "-", length: Optional[int] = None):
"""打印分隔线"""
if length is None:
try:
import shutil
length = shutil.get_terminal_size().columns
except:
length = 80
line_str = char * min(length, 100) # 限制最大长度
if self.color_enabled:
print(f"{Colors.BRIGHT_BLACK}{line_str}{Colors.RESET}")
else:
print(line_str)
def blank(self):
"""打印空行"""
print()
# 创建默认日志实例
default_logger = Logs()
# 为了方便使用,提供模块级别的函数
def debug(msg: str):
default_logger.debug(msg)
def info(msg: str):
default_logger.info(msg)
def ok(msg: str):
default_logger.ok(msg)
def warn(msg: str):
default_logger.warn(msg)
def error(msg: str):
default_logger.error(msg)
def critical(msg: str):
default_logger.critical(msg)
# 使用示例
if __name__ == "__main__":
# 示例1基本使用
print("=" * 60)
print("示例1:基本使用")
print("=" * 60)
log = Logs(name="Test", show_time=True, show_level=True)
log.ok("操作成功完成")
log.info("正在加载配置...")
log.warn("配置文件未找到,使用默认配置")
log.error("连接数据库失败")
log.debug("这是调试信息") # 默认 INFO 级别,不会显示
print("\n")
# 示例2使用 __call__ 方法self.log('msg', 'err')
print("=" * 60)
print("示例2:使用 __call__ 方法")
print("=" * 60)
log2 = Logs(name="MyApp", show_time=False)
log2("这是一条普通信息", "info")
log2("这是一条警告", "warn")
log2("这是一条错误", "err")
log2("这是一条成功消息", "ok")
log2("这是调试信息", "debug")
print("\n")
# 示例3固定颜色
print("=" * 60)
print("示例3固定颜色输出")
print("=" * 60)
log3 = Logs(name="ColorTest")
log3("这条消息是红色的", "error", "red")
log3("这条消息是绿色的", "info", "green")
log3("这条消息是黄色的", "warn", "yellow")
log3("这条消息是蓝色的", "ok", "blue")
print("\n")
# 示例4不使用颜色
print("=" * 60)
print("示例4禁用颜色")
print("=" * 60)
log4 = Logs(name="NoColor", color_enabled=False)
log4.ok("没有颜色的成功消息")
log4.error("没有颜色的错误消息")
print("\n")
# 示例5分隔线
print("=" * 60)
print("示例5: 工具方法")
print("=" * 60)
log5 = Logs(name="Tool")
log5.line("=")
log5.ok("功能演示")
log5.line("-")
log5.blank()
log5.info("完成")
log5.line("=")

View File

@@ -6,9 +6,16 @@ import sunhpc
import signal import signal
import syslog import syslog
import sunhpc.util import sunhpc.util
import sunhpc.error as error
from sunhpc.util import CommandError
from sunhpc.logs import Logs
slog = Logs(name="Sunhpc", show_time=True)
from sunhpc.util import getTerminalWidth
def _main(argv=None): def _main(argv=None):
try: try:
database = None database = None
except ImportError: except ImportError:
@@ -46,7 +53,7 @@ def _main(argv=None):
print ('error - invalid sunhpc command "%s"' % args[0]) print ('error - invalid sunhpc command "%s"' % args[0])
sys.exit(-1) sys.exit(-1)
name = ' '.join('.'.split(s)) name = ' '.join(s.split('.')[2:])
try: try:
command = getattr(module, 'Command')(database) command = getattr(module, 'Command')(database)
@@ -63,25 +70,36 @@ def _main(argv=None):
command.isRootUser() or command.isApacheUser()): command.isRootUser() or command.isApacheUser()):
os.system('sudo %s' % ' '.join(sys.argv)) os.system('sudo %s' % ' '.join(sys.argv))
else: else:
#import sunhpc
try: try:
command.runWrapper(name, args[i:]) command.runWrapper(name, args[i:])
text = command.getText() text = command.getText()
if len(text) > 0: if len(text) > 0:
print (text, end=' ') if text.endswith('\n'):
if text[len(text)-1] != '\n': sys.stdout.write(text)
else:
print (text)
print () print ()
except sunhpc.util.CommandError as e: except CommandError as e:
print ('Error: ', e) print ('Error: ', e)
print (command.usage()) print (command.usage())
exit(1) return 1
config_file = '/etc/sunhpc/sunhpc.yaml'
if not os.path.exists(config_file):
slog.error('=' * (getTerminalWidth() - 30))
slog.error ('config file "%s" not found' % config_file)
slog.info('Run following command to create the config file:')
slog.info(' # sunhpc init config eth1')
slog.error('=' * (getTerminalWidth() - 30))
syslog.closelog() syslog.closelog()
return 0
def main(argv=None): def main(argv=None):
if any(flag in argv for flag in ['--debug', '-d']):
_main(argv) _main(argv)
''' else:
try: try:
return _main(argv) return _main(argv)
except KeyboardInterrupt: except KeyboardInterrupt:
@@ -89,10 +107,14 @@ def main(argv=None):
tty.error("Keyboard interrupt.") tty.error("Keyboard interrupt.")
return signal.SIGINT.value return signal.SIGINT.value
except AttributeError:
sys.stderr.write("\n")
tty.error("Attribute error.")
return 1
except SystemExit as e: except SystemExit as e:
return e.code return e.code
except Exception as e: except Exception as e:
tty.error(e) tty.error(e)
return 3 return 3
'''

880
lib/sunhpc/sunhpc/memory.py Normal file
View File

@@ -0,0 +1,880 @@
"""
MemoryInfo class to retrieve comprehensive system memory information.
Supports physical RAM, swap memory, and virtualization detection.
No third-party libraries required, uses /proc, /sys, and system commands.
"""
import os
import re
import glob
from typing import Optional, Dict, List, Any, Tuple
class MemoryInfo:
"""Class to retrieve detailed information about system memory."""
def __init__(self):
self._info_cache: Dict[str, Any] = {}
def get_memory_info(self, refresh: bool = False) -> Dict[str, Any]:
"""
Get comprehensive memory information.
Returns dictionary with all available memory information.
"""
if not refresh and self._info_cache:
return self._info_cache
info = {
'physical_memory': self._get_physical_memory_info(),
'swap_memory': self._get_swap_info(),
'virtual_memory': self._get_virtual_memory_stats(),
'memory_modules': self._get_memory_modules_info(),
'top_memory_processes': self._get_top_memory_processes(),
'memory_stats': self._get_memory_statistics(),
'numa_info': self._get_numa_info(),
'is_virtual': False,
'virtual_type': None
}
# Detect virtualization
self._detect_virtualization(info)
self._info_cache = info
return info
def _get_physical_memory_info(self) -> Dict[str, Any]:
"""Get physical memory information from /proc/meminfo and /sys."""
mem_info = {
'total_bytes': None,
'total_human': None,
'available_bytes': None,
'available_human': None,
'free_bytes': None,
'free_human': None,
'used_bytes': None,
'used_human': None,
'used_percent': None,
'buffers_bytes': None,
'cached_bytes': None,
'shared_bytes': None,
'slab_bytes': None,
'hugepages_total': None,
'hugepages_free': None,
'hugepages_size_bytes': None,
'hugepages_size_human': None
}
# Parse /proc/meminfo
try:
with open('/proc/meminfo', 'r') as f:
mem_data = f.read()
# Extract values
patterns = {
'total_bytes': r'MemTotal:\s+(\d+)\s+kB',
'available_bytes': r'MemAvailable:\s+(\d+)\s+kB',
'free_bytes': r'MemFree:\s+(\d+)\s+kB',
'buffers_bytes': r'Buffers:\s+(\d+)\s+kB',
'cached_bytes': r'Cached:\s+(\d+)\s+kB',
'shared_bytes': r'Shmem:\s+(\d+)\s+kB',
'slab_bytes': r'Slab:\s+(\d+)\s+kB',
'hugepages_total': r'HugePages_Total:\s+(\d+)',
'hugepages_free': r'HugePages_Free:\s+(\d+)',
'hugepages_size_bytes': r'Hugepagesize:\s+(\d+)\s+kB'
}
for key, pattern in patterns.items():
match = re.search(pattern, mem_data, re.MULTILINE)
if match:
value = int(match.group(1))
if key.endswith('_bytes'):
# Convert KB to bytes
value *= 1024
mem_info[key] = value
# Calculate used memory
if mem_info['total_bytes'] and mem_info['available_bytes']:
mem_info['used_bytes'] = mem_info['total_bytes'] - mem_info['available_bytes']
mem_info['used_human'] = self._bytes_to_human(mem_info['used_bytes'])
mem_info['used_percent'] = (mem_info['used_bytes'] / mem_info['total_bytes']) * 100
# Convert bytes to human readable
for key in ['total_bytes', 'available_bytes', 'free_bytes', 'buffers_bytes',
'cached_bytes', 'shared_bytes', 'slab_bytes']:
if mem_info.get(key):
human_key = key.replace('_bytes', '_human')
mem_info[human_key] = self._bytes_to_human(mem_info[key])
if mem_info.get('hugepages_size_bytes'):
mem_info['hugepages_size_human'] = self._bytes_to_human(mem_info['hugepages_size_bytes'])
except Exception as e:
pass
return mem_info
def _get_swap_info(self) -> Dict[str, Any]:
"""Get swap memory information."""
swap_info = {
'total_bytes': None,
'total_human': None,
'free_bytes': None,
'free_human': None,
'used_bytes': None,
'used_human': None,
'used_percent': None,
'swap_devices': []
}
# Parse /proc/meminfo for swap totals
try:
with open('/proc/meminfo', 'r') as f:
mem_data = f.read()
total_match = re.search(r'SwapTotal:\s+(\d+)\s+kB', mem_data)
free_match = re.search(r'SwapFree:\s+(\d+)\s+kB', mem_data)
if total_match:
swap_info['total_bytes'] = int(total_match.group(1)) * 1024
swap_info['total_human'] = self._bytes_to_human(swap_info['total_bytes'])
if free_match:
swap_info['free_bytes'] = int(free_match.group(1)) * 1024
swap_info['free_human'] = self._bytes_to_human(swap_info['free_bytes'])
if swap_info['total_bytes'] and swap_info['free_bytes']:
swap_info['used_bytes'] = swap_info['total_bytes'] - swap_info['free_bytes']
swap_info['used_human'] = self._bytes_to_human(swap_info['used_bytes'])
if swap_info['total_bytes'] > 0:
swap_info['used_percent'] = (swap_info['used_bytes'] / swap_info['total_bytes']) * 100
except Exception:
pass
# Get individual swap devices from /proc/swaps
try:
with open('/proc/swaps', 'r') as f:
lines = f.readlines()[1:] # Skip header
for line in lines:
parts = line.strip().split()
if len(parts) >= 4:
swap_dev = {
'device': parts[0],
'type': parts[1],
'size_bytes': int(parts[2]) * 1024,
'size_human': self._bytes_to_human(int(parts[2]) * 1024),
'used_bytes': int(parts[3]) * 1024,
'used_human': self._bytes_to_human(int(parts[3]) * 1024),
'priority': int(parts[4]) if len(parts) > 4 else None
}
swap_info['swap_devices'].append(swap_dev)
except Exception:
pass
return swap_info
def _get_virtual_memory_stats(self) -> Dict[str, Any]:
"""Get virtual memory statistics from /proc/vmstat."""
vm_stats = {
'pgpgin': None, # Pages paged in
'pgpgout': None, # Pages paged out
'pswpin': None, # Pages swapped in
'pswpout': None, # Pages swapped out
'pgfault': None, # Page faults
'pgmajfault': None, # Major page faults
'pgfree': None, # Pages freed
'pgactivate': None, # Pages activated
'pgdeactivate': None,# Pages deactivated
'pglaundry': None, # Pages laundried
'pgrefill': None, # Pages refilled
'pgsteal': None, # Pages stolen
'pgscan_kswapd': None, # Pages scanned by kswapd
'pgscan_direct': None, # Pages scanned directly
'pgscan_khugepaged': None, # Pages scanned by khugepaged
}
try:
with open('/proc/vmstat', 'r') as f:
for line in f:
parts = line.strip().split()
if len(parts) == 2:
key, value = parts
if key in vm_stats:
vm_stats[key] = int(value)
except Exception:
pass
return vm_stats
def _get_memory_modules_info(self) -> List[Dict[str, Any]]:
"""
Get physical memory module information from DMIDECODE or /sys.
Requires root for full details.
"""
modules = []
# Try to get from /sys/devices/system/edac/mc/
try:
mc_paths = glob.glob('/sys/devices/system/edac/mc/mc*')
for mc_path in mc_paths:
dimm_paths = glob.glob(f'{mc_path}/dimm/dimm*')
for dimm_path in dimm_paths:
module = self._read_dimm_info(dimm_path)
if module:
modules.append(module)
except Exception:
pass
# Try dmidecode for detailed module info (requires root)
if not modules:
modules = self._get_dmidecode_memory_info()
# If no module info found, try to get from /proc/meminfo
if not modules:
modules = self._get_basic_memory_info()
return modules
def _read_dimm_info(self, dimm_path: str) -> Optional[Dict[str, Any]]:
"""Read DIMM information from sysfs."""
module = {}
try:
# Try to read various attributes
attrs = {
'size': 'size',
'manufacturer': 'manufacturer',
'part_number': 'part_num',
'serial_number': 'serial',
'rank': 'rank'
}
for attr_file, attr_name in attrs.items():
file_path = f'{dimm_path}/{attr_file}'
if os.path.exists(file_path):
with open(file_path, 'r') as f:
value = f.read().strip()
if value and not value.isspace():
module[attr_name] = value
# Parse size (usually in MB)
if 'size' in module:
try:
size_mb = int(module['size'])
module['size_bytes'] = size_mb * 1024 * 1024
module['size_human'] = self._bytes_to_human(module['size_bytes'])
except ValueError:
pass
if module:
module['type'] = 'Unknown'
return module
except Exception:
pass
return None
def _get_dmidecode_memory_info(self) -> List[Dict[str, Any]]:
"""Get detailed memory module info using dmidecode command."""
modules = []
if not self._check_command_exists('dmidecode'):
return modules
try:
# Run dmidecode to get memory information
result = self._run_command(['dmidecode', '-t', 'memory'], timeout=10)
if not result:
return modules
# Parse dmidecode output
current_module = {}
in_memory_device = False
for line in result.split('\n'):
line = line.strip()
if 'Memory Device' in line:
if current_module and in_memory_device:
modules.append(current_module)
current_module = {}
in_memory_device = True
elif in_memory_device and ':' in line:
key, value = line.split(':', 1)
key = key.strip().lower().replace(' ', '_')
value = value.strip()
if key == 'size':
if value != 'No Module Installed' and 'MB' in value:
size_mb = int(value.split()[0])
current_module['size_bytes'] = size_mb * 1024 * 1024
current_module['size_human'] = self._bytes_to_human(current_module['size_bytes'])
current_module['size_mb'] = size_mb
elif key == 'Locator':
current_module['locator'] = value
elif key == 'type':
current_module['type'] = value
elif key == 'speed':
current_module['speed'] = value
elif key == 'manufacturer':
current_module['manufacturer'] = value
elif key == 'serial_number':
current_module['serial'] = value
elif key == 'asset_tag':
current_module['asset_tag'] = value
elif key == 'part_number':
current_module['part_number'] = value
elif key == 'rank':
current_module['rank'] = value
elif key == 'configured_clock_speed':
current_module['configured_speed'] = value
elif key == 'configured_memory_speed':
current_module['configured_memory_speed'] = value
elif key == 'minimum_voltage':
current_module['minimum_voltage'] = value
elif key == 'maximum_voltage':
current_module['maximum_voltage'] = value
elif key == 'configured_voltage':
current_module['configured_voltage'] = value
elif key == 'memory_technology':
current_module['memory_technology'] = value
elif key == 'type_detail':
current_module['type_detail'] = value
if current_module and in_memory_device:
modules.append(current_module)
except Exception:
pass
return modules
def _get_basic_memory_info(self) -> List[Dict[str, Any]]:
"""Get basic memory information when detailed info is not available."""
modules = []
# Get total memory size
total_bytes = None
try:
with open('/proc/meminfo', 'r') as f:
for line in f:
if 'MemTotal' in line:
total_kb = int(re.search(r'(\d+)', line).group(1))
total_bytes = total_kb * 1024
break
except Exception:
pass
if total_bytes:
module = {
'size_bytes': total_bytes,
'size_human': self._bytes_to_human(total_bytes),
'type': 'Unknown',
'manufacturer': 'Unknown',
'serial': 'Unknown',
'part_number': 'Unknown',
'note': 'Detailed information not available (requires root or dmidecode)'
}
modules.append(module)
return modules
def _get_top_memory_processes(self, count: int = 10) -> List[Dict[str, Any]]:
"""Get top memory-consuming processes."""
processes = []
try:
# Read /proc for process information
for pid in os.listdir('/proc'):
if not pid.isdigit():
continue
try:
# Read process status
with open(f'/proc/{pid}/status', 'r') as f:
status = f.read()
# Get process name
name_match = re.search(r'Name:\s+(.+?)\n', status)
if not name_match:
continue
name = name_match.group(1)
# Get memory usage (VmRSS in kB)
vmrss_match = re.search(r'VmRSS:\s+(\d+)\s+kB', status)
if not vmrss_match:
continue
memory_kb = int(vmrss_match.group(1))
if memory_kb > 0:
processes.append({
'pid': int(pid),
'name': name,
'memory_kb': memory_kb,
'memory_bytes': memory_kb * 1024,
'memory_human': self._bytes_to_human(memory_kb * 1024)
})
except (IOError, OSError, ValueError):
continue
except Exception:
pass
# Sort by memory usage (descending) and return top N
processes.sort(key=lambda x: x['memory_kb'], reverse=True)
return processes[:count]
def _get_memory_statistics(self) -> Dict[str, Any]:
"""Get memory statistics and pressure information."""
stats = {
'load_average': None,
'memory_pressure': None,
'swap_pressure': None,
'io_pressure': None,
'oom_killer_enabled': None,
'memory_fragmentation': None
}
# Get load average
try:
with open('/proc/loadavg', 'r') as f:
load = f.read().strip().split()
if len(load) >= 3:
stats['load_average'] = {
'1min': float(load[0]),
'5min': float(load[1]),
'15min': float(load[2])
}
except Exception:
pass
# Check if OOM killer is enabled
try:
with open('/proc/sys/vm/oom_kill_allocating_task', 'r') as f:
stats['oom_killer_enabled'] = int(f.read().strip()) == 1
except Exception:
pass
# Memory pressure (if using cgroup v2)
try:
if os.path.exists('/proc/pressure/memory'):
with open('/proc/pressure/memory', 'r') as f:
content = f.read().strip()
stats['memory_pressure'] = self._parse_pressure_info(content)
if os.path.exists('/proc/pressure/swap'):
with open('/proc/pressure/swap', 'r') as f:
content = f.read().strip()
stats['swap_pressure'] = self._parse_pressure_info(content)
if os.path.exists('/proc/pressure/io'):
with open('/proc/pressure/io', 'r') as f:
content = f.read().strip()
stats['io_pressure'] = self._parse_pressure_info(content)
except Exception:
pass
# Memory fragmentation info
try:
with open('/proc/buddyinfo', 'r') as f:
stats['memory_fragmentation'] = f.read().strip()
except Exception:
pass
return stats
def _get_numa_info(self) -> Optional[Dict[str, Any]]:
"""Get NUMA (Non-Uniform Memory Access) information."""
numa_info = {
'numa_enabled': False,
'numa_nodes': [],
'numa_stats': {}
}
# Check if NUMA is enabled
if not os.path.exists('/sys/devices/system/node'):
return None
try:
# Get NUMA nodes
node_paths = glob.glob('/sys/devices/system/node/node*')
if node_paths:
numa_info['numa_enabled'] = True
for node_path in node_paths:
node_id = os.path.basename(node_path).replace('node', '')
node_info = {
'node_id': int(node_id),
'memory_bytes': None,
'memory_human': None,
'cpus': []
}
# Get node memory
meminfo_path = f'{node_path}/meminfo'
if os.path.exists(meminfo_path):
with open(meminfo_path, 'r') as f:
for line in f:
if 'MemTotal' in line:
mem_kb = int(re.search(r'(\d+)', line).group(1))
node_info['memory_bytes'] = mem_kb * 1024
node_info['memory_human'] = self._bytes_to_human(node_info['memory_bytes'])
break
# Get CPUs in this node
cpulist_path = f'{node_path}/cpulist'
if os.path.exists(cpulist_path):
with open(cpulist_path, 'r') as f:
cpu_list = f.read().strip()
node_info['cpus'] = self._parse_cpu_list(cpu_list)
numa_info['numa_nodes'].append(node_info)
# Get NUMA statistics
if os.path.exists('/proc/zoneinfo'):
with open('/proc/zoneinfo', 'r') as f:
content = f.read()
# Parse node-specific stats
for node in numa_info['numa_nodes']:
node_id = node['node_id']
pattern = rf'Node {node_id}.*?pages free\s+(\d+)'
match = re.search(pattern, content, re.DOTALL)
if match:
numa_info['numa_stats'][f'node_{node_id}_free_pages'] = int(match.group(1))
except Exception:
pass
return numa_info
def _detect_virtualization(self, info: Dict[str, Any]):
"""Detect if system is virtualized."""
# Check CPU info
try:
with open('/proc/cpuinfo', 'r') as f:
cpuinfo = f.read().lower()
if 'hypervisor' in cpuinfo:
info['is_virtual'] = True
if 'kvm' in cpuinfo:
info['virtual_type'] = 'KVM'
elif 'vmware' in cpuinfo:
info['virtual_type'] = 'VMware'
elif 'virtualbox' in cpuinfo:
info['virtual_type'] = 'VirtualBox'
elif 'xen' in cpuinfo:
info['virtual_type'] = 'Xen'
else:
info['virtual_type'] = 'Unknown'
except Exception:
pass
# Check DMI info
if not info['is_virtual']:
try:
if os.path.exists('/sys/class/dmi/id/product_name'):
with open('/sys/class/dmi/id/product_name', 'r') as f:
product = f.read().strip().lower()
if 'vmware' in product:
info['is_virtual'] = True
info['virtual_type'] = 'VMware'
elif 'virtualbox' in product:
info['is_virtual'] = True
info['virtual_type'] = 'VirtualBox'
elif 'kvm' in product or 'qemu' in product:
info['is_virtual'] = True
info['virtual_type'] = 'KVM/QEMU'
except Exception:
pass
@staticmethod
def _parse_pressure_info(content: str) -> Dict[str, Any]:
"""Parse pressure stall information."""
pressure = {}
for line in content.split('\n'):
if line.startswith('some'):
parts = line.split()
pressure['some'] = {
'avg10': float(parts[1].split('=')[1]),
'avg60': float(parts[2].split('=')[1]),
'avg300': float(parts[3].split('=')[1]),
'total': int(parts[4].split('=')[1])
}
elif line.startswith('full'):
parts = line.split()
pressure['full'] = {
'avg10': float(parts[1].split('=')[1]),
'avg60': float(parts[2].split('=')[1]),
'avg300': float(parts[3].split('=')[1]),
'total': int(parts[4].split('=')[1])
}
return pressure
@staticmethod
def _parse_cpu_list(cpu_list: str) -> List[int]:
"""Parse CPU list string (e.g., '0-3,5,7-9') into list of CPU numbers."""
cpus = []
for part in cpu_list.split(','):
if '-' in part:
start, end = map(int, part.split('-'))
cpus.extend(range(start, end + 1))
else:
cpus.append(int(part))
return cpus
def get_memory_summary(self) -> str:
"""Get a human-readable summary of memory information."""
info = self.get_memory_info()
lines = []
lines.append("=" * 80)
lines.append("MEMORY INFORMATION SUMMARY")
lines.append("=" * 80)
# Physical Memory
phys = info['physical_memory']
lines.append("\nPhysical Memory:")
lines.append(f" Total: {phys.get('total_human', 'Unknown')}")
lines.append(f" Used: {phys.get('used_human', 'Unknown')} ({phys.get('used_percent', 0):.1f}%)")
lines.append(f" Available: {phys.get('available_human', 'Unknown')}")
lines.append(f" Free: {phys.get('free_human', 'Unknown')}")
lines.append(f" Buffers: {phys.get('buffers_human', 'Unknown')}")
lines.append(f" Cached: {phys.get('cached_human', 'Unknown')}")
if phys.get('hugepages_total', 0) > 0:
lines.append(f" HugePages: {phys['hugepages_total']} total, {phys['hugepages_free']} free")
lines.append(f" HugePage Size: {phys.get('hugepages_size_human', 'Unknown')}")
# Swap Memory
swap = info['swap_memory']
if swap.get('total_bytes', 0) > 0:
lines.append("\nSwap Memory:")
lines.append(f" Total: {swap.get('total_human', 'Unknown')}")
lines.append(f" Used: {swap.get('used_human', 'Unknown')} ({swap.get('used_percent', 0):.1f}%)")
lines.append(f" Free: {swap.get('free_human', 'Unknown')}")
if swap.get('swap_devices'):
lines.append(" Swap Devices:")
for dev in swap['swap_devices']:
lines.append(f" - {dev['device']} ({dev['type']}): {dev['size_human']}, "
f"used: {dev['used_human']}")
# Memory Modules
modules = info['memory_modules']
if modules:
lines.append(f"\nMemory Modules ({len(modules)}):")
for i, module in enumerate(modules, 1):
lines.append(f" Module {i}:")
if 'size_human' in module:
lines.append(f" Size: {module['size_human']}")
if 'type' in module:
lines.append(f" Type: {module['type']}")
if 'speed' in module:
lines.append(f" Speed: {module['speed']}")
if 'manufacturer' in module and module['manufacturer'] != 'Unknown':
lines.append(f" Manufacturer: {module['manufacturer']}")
if 'part_number' in module and module['part_number'] != 'Unknown':
lines.append(f" Part Number: {module['part_number']}")
if 'serial' in module and module['serial'] != 'Unknown':
lines.append(f" Serial: {module['serial']}")
if 'note' in module:
lines.append(f" Note: {module['note']}")
if 'configured_voltage' in module:
lines.append(f" Configured Voltage: {module['configured_voltage']}")
if 'memory_technology' in module:
lines.append(f" Memory Technology: {module['memory_technology']}")
if 'type_detail' in module:
lines.append(f" Type Detail: {module['type_detail']}")
if 'minimum_voltage' in module:
lines.append(f" Minimum Voltage: {module['minimum_voltage']}")
if 'maximum_voltage' in module:
lines.append(f" Maximum Voltage: {module['maximum_voltage']}")
if 'location' in module:
lines.append(f" Location: {module['location']}")
# Top Memory Processes
processes = info['top_memory_processes']
if processes:
lines.append(f"\nTop {len(processes)} Memory-Consuming Processes:")
for proc in processes:
lines.append(f" PID {proc['pid']:6d} - {proc['name']:20s}: {proc['memory_human']}")
# Virtualization
if info['is_virtual']:
lines.append(f"\nVirtualization: {info['virtual_type']}")
# Load Average
stats = info['memory_stats']
if stats.get('load_average'):
la = stats['load_average']
lines.append(f"\nLoad Average: {la['1min']:.2f} (1min), {la['5min']:.2f} (5min), {la['15min']:.2f} (15min)")
# Memory Pressure (if available)
if stats.get('memory_pressure'):
lines.append("\nMemory Pressure:")
pressure = stats['memory_pressure']
if 'some' in pressure:
lines.append(f" Some: avg10={pressure['some']['avg10']:.2f}, avg60={pressure['some']['avg60']:.2f}, avg300={pressure['some']['avg300']:.2f}")
lines.append("\n" + "=" * 80)
return '\n'.join(lines)
def get_memory_usage_warning(self, threshold_percent: float = 90.0) -> Optional[str]:
"""
Check if memory usage exceeds threshold and return warning.
threshold_percent: Threshold percentage for warning (default 90%)
"""
info = self.get_memory_info()
phys = info['physical_memory']
if phys.get('used_percent') and phys['used_percent'] >= threshold_percent:
return (f"WARNING: Memory usage is {phys['used_percent']:.1f}% "
f"({phys['used_human']} used of {phys['total_human']} total)")
swap = info['swap_memory']
if swap.get('used_percent') and swap.get('total_bytes', 0) > 0:
if swap['used_percent'] >= threshold_percent:
return (f"WARNING: Swap usage is {swap['used_percent']:.1f}% "
f"({swap['used_human']} used of {swap['total_human']} total)")
return None
def refresh(self) -> 'MemoryInfo':
"""Clear cache and refresh memory information."""
self._info_cache = {}
return self
@staticmethod
def _bytes_to_human(bytes_val: Optional[int]) -> str:
"""Convert bytes to human-readable format."""
if bytes_val is None:
return "Unknown"
for unit in ['B', 'KB', 'MB', 'GB', 'TB', 'PB']:
if bytes_val < 1024.0:
return f"{bytes_val:.2f} {unit}"
bytes_val /= 1024.0
return f"{bytes_val:.2f} EB"
@staticmethod
def _check_command_exists(command: str) -> bool:
"""Check if a command exists in the system."""
import subprocess
try:
subprocess.run(['which', command], capture_output=True, timeout=5)
return True
except Exception:
return False
@staticmethod
def _run_command(command: List[str], timeout: int = 10) -> Optional[str]:
"""Run a command and return its output."""
import subprocess
try:
result = subprocess.run(
command,
capture_output=True,
text=True,
timeout=timeout
)
return result.stdout if result.returncode == 0 else None
except Exception:
return None
def __repr__(self) -> str:
info = self.get_memory_info()
phys = info['physical_memory']
total = phys.get('total_human', 'Unknown')
used_percent = phys.get('used_percent', 0)
return f"MemoryInfo(total={total}, used={used_percent:.1f}%)"
# ------------------------------------------------------------------
# Example usage and test
# ------------------------------------------------------------------
if __name__ == '__main__':
memory = MemoryInfo()
# Print summary
print(memory.get_memory_summary())
# Get detailed information as dictionary
print("\n" + "=" * 80)
print("DETAILED INFORMATION (JSON-like structure)")
print("=" * 80)
info = memory.get_memory_info()
print("\nPhysical Memory:")
phys = info['physical_memory']
for key, value in phys.items():
if value is not None:
print(f" {key}: {value}")
print("\nSwap Memory:")
swap = info['swap_memory']
for key, value in swap.items():
if value is not None and key != 'swap_devices':
print(f" {key}: {value}")
if swap.get('swap_devices'):
print("\n Swap Devices:")
for dev in swap['swap_devices']:
print(f" - {dev}")
print("\nVirtual Memory Statistics:")
vm_stats = info['virtual_memory']
for key, value in vm_stats.items():
if value is not None:
print(f" {key}: {value:,}")
if info['numa_info']:
print("\nNUMA Information:")
numa = info['numa_info']
print(f" NUMA Enabled: {numa['numa_enabled']}")
if numa['numa_nodes']:
print(f" NUMA Nodes: {len(numa['numa_nodes'])}")
for node in numa['numa_nodes']:
print(f" Node {node['node_id']}: {node.get('memory_human', 'Unknown')}, "
f"CPUs: {node['cpus'][:10]}{'...' if len(node['cpus']) > 10 else ''}")
# Check for warnings
warning = memory.get_memory_usage_warning(threshold_percent=80)
if warning:
print(f"\n{warning}")
else:
print("\nMemory usage is within normal limits")
# Monitor changes over time (example)
print("\n" + "=" * 80)
print("MONITORING MEMORY CHANGES (5 second interval)")
print("=" * 80)
import time
for i in range(3):
phys = memory.get_memory_info()['physical_memory']
print(f"\nSample {i+1}: Used: {phys['used_human']} ({phys['used_percent']:.1f}%), "
f"Available: {phys['available_human']}")
if i < 2:
time.sleep(5)
memory.refresh() # Refresh data for next sample

View File

@@ -0,0 +1,523 @@
"""
Network class to retrieve network interface information and manipulate IP addresses.
No third-party libraries required, uses /sys filesystem and ioctl system calls.
"""
import os
import re
import struct
import socket
import fcntl
from typing import Optional, Tuple, List, Union
class Network:
def __init__(self):
self.iface: Optional[str] = None
self._ip: Optional[str] = None
self._netmask: Optional[str] = None
self._network: Optional[str] = None
self._gateway: Optional[str] = None
self._mac: Optional[str] = None
self._bytes_rx: Optional[int] = None
self._bytes_tx: Optional[int] = None
self._cidr: Optional[int] = None
self._ipv6: Optional[List[str]] = None
self._manual_ip: bool = False # Flag to track if IP was manually set
self._manual_netmask: bool = False # Flag to track if netmask was manually set
def setiface(self, iface: str) -> 'Network':
"""Set the network interface and refresh all information."""
self.iface = iface
# Reset manual flags when changing interface
self._manual_ip = False
self._manual_netmask = False
self._refresh_all()
return self
def setip(self, ip: str) -> 'Network':
"""Manually set the IPv4 address (overrides system value)."""
self._ip = ip
self._manual_ip = True
# Update dependent values
if self._manual_netmask:
self._update_dependent_values()
else:
# Only update network and cidr if we have netmask
self._network = self.getnetwork()
self._cidr = self.getcidr()
return self
def setnetmask(self, netmask: str) -> 'Network':
"""Manually set the netmask (overrides system value)."""
self._netmask = netmask
self._manual_netmask = True
# Update dependent values
if self._manual_ip:
self._update_dependent_values()
else:
# Only update network and cidr if we have IP
self._network = self.getnetwork()
self._cidr = self.getcidr()
return self
def _update_dependent_values(self):
"""Update network and CIDR based on current IP and netmask."""
if self._ip and self._netmask:
self._network = self._calculate_network(self._ip, self._netmask)
self._cidr = self._calculate_cidr(self._netmask)
else:
self._network = None
self._cidr = None
def _refresh_all(self):
"""Refresh all interface information respecting manual overrides."""
if not self.iface:
return
# Only get from system if not manually set
if not self._manual_ip:
self._ip = self.getip()
if not self._manual_netmask:
self._netmask = self.getnetmask()
# Always refresh these from system (they can't be manually set)
self._gateway = self.getgateway()
self._mac = self.getmac()
self._bytes_rx, self._bytes_tx = self.get_transfer_bytes()
self._ipv6 = self.getipv6()
# Update dependent values
self._update_dependent_values()
# ------------------------------------------------------------------
# Public getter methods
# ------------------------------------------------------------------
def getip(self) -> Optional[str]:
"""Get IPv4 address of the interface."""
if self._manual_ip:
return self._ip
return self._get_ip_from_ioctl()
def getnetmask(self) -> Optional[str]:
"""Get netmask of the interface."""
if self._manual_netmask:
return self._netmask
return self._get_netmask_from_ioctl()
def getnetwork(self) -> Optional[str]:
"""Calculate network address from IP and netmask."""
if self._network is not None:
return self._network
ip = self.getip()
mask = self.getnetmask()
return self._calculate_network(ip, mask)
def _calculate_network(self, ip: Optional[str], mask: Optional[str]) -> Optional[str]:
"""Calculate network address from IP and netmask."""
if not ip or not mask:
return None
try:
ip_int = self._ip_to_int(ip)
mask_int = self._ip_to_int(mask)
network_int = ip_int & mask_int
return self._int_to_ip(network_int)
except Exception:
return None
def getgateway(self) -> Optional[str]:
"""Get default gateway for the interface."""
return self._get_gateway_from_proc()
def getmac(self) -> Optional[str]:
"""Get MAC address of the interface."""
return self._get_mac_from_sys()
def get_transfer_bytes(self) -> Tuple[Optional[int], Optional[int]]:
"""Get total received and transmitted bytes (raw integers)."""
rx, tx = self._get_stats_from_sys()
return rx, tx
def get_transfer_human(self) -> Tuple[Optional[str], Optional[str]]:
"""Get total received and transmitted bytes in human-readable format."""
rx, tx = self.get_transfer_bytes()
return self._bytes_to_human(rx), self._bytes_to_human(tx)
def getcidr(self) -> Optional[int]:
"""Get CIDR prefix length from netmask."""
if self._cidr is not None:
return self._cidr
mask = self.getnetmask()
return self._calculate_cidr(mask)
def _calculate_cidr(self, mask: Optional[str]) -> Optional[int]:
"""Calculate CIDR from netmask."""
if not mask:
return None
try:
mask_int = self._ip_to_int(mask)
# Count leading 1 bits
cidr = bin(mask_int).count('1')
return cidr if 0 <= cidr <= 32 else None
except Exception:
return None
def getipv6(self) -> Optional[List[str]]:
"""Get all IPv6 addresses of the interface (link-local and global)."""
return self._get_ipv6_from_proc()
# ------------------------------------------------------------------
# IP iteration (inc parameter)
# ------------------------------------------------------------------
def iter_ip(self, inc: int = 1, ip: Optional[str] = None, netmask: Optional[str] = None) -> Optional[str]:
"""
Return next or previous IP address based on increment.
Args:
inc: Increment step (positive for forward, negative for backward)
ip: Optional custom IP address (uses current IP if not provided)
netmask: Optional custom netmask (uses current netmask if not provided)
Examples:
net.iter_ip(1) # Use current IP and netmask
net.iter_ip(-1, '10.1.1.1', '255.255.255.0') # Custom IP and netmask
net.iter_ip(2, ip='192.168.1.1', netmask='24') # CIDR notation supported
"""
# Use provided values or fall back to current settings
use_ip = ip if ip is not None else self.getip()
use_netmask = netmask if netmask is not None else self.getnetmask()
# Convert CIDR notation to netmask if needed
if use_netmask and isinstance(use_netmask, str) and '/' not in use_netmask:
if use_netmask.isdigit() and 0 <= int(use_netmask) <= 32:
use_netmask = self._cidr_to_netmask(int(use_netmask))
if not use_ip or not use_netmask:
return None
try:
ip_int = self._ip_to_int(use_ip)
mask_int = self._ip_to_int(use_netmask)
network_int = ip_int & mask_int
broadcast_int = network_int | (~mask_int & 0xFFFFFFFF)
# Calculate new IP
new_ip_int = ip_int + inc
# Clamp to network range (exclude network and broadcast addresses)
if new_ip_int <= network_int:
# Wrap to the end of network (minus broadcast)
new_ip_int = broadcast_int - 1
elif new_ip_int >= broadcast_int:
# Wrap to the start of network (plus network address)
new_ip_int = network_int + 1
# Ensure we don't return network or broadcast address
if new_ip_int == network_int:
new_ip_int = broadcast_int - 1
elif new_ip_int == broadcast_int:
new_ip_int = network_int + 1
return self._int_to_ip(new_ip_int)
except Exception:
return None
def iter_ip_range(self, start_ip: str, netmask: str, steps: int = 10) -> List[str]:
"""
Generate a range of IP addresses.
Args:
start_ip: Starting IP address
netmask: Netmask (dotted decimal or CIDR)
steps: Number of steps to generate (can be positive or negative)
Returns:
List of IP addresses
"""
results = []
for i in range(1, abs(steps) + 1):
step = i if steps > 0 else -i
next_ip = self.iter_ip(step, start_ip, netmask)
if next_ip:
results.append(next_ip)
return results
@staticmethod
def _cidr_to_netmask(cidr: int) -> str:
"""Convert CIDR prefix length to dotted netmask."""
mask_int = (0xFFFFFFFF << (32 - cidr)) & 0xFFFFFFFF
return f"{(mask_int >> 24) & 0xFF}.{(mask_int >> 16) & 0xFF}.{(mask_int >> 8) & 0xFF}.{mask_int & 0xFF}"
# ------------------------------------------------------------------
# Low-level implementation methods (with fallbacks)
# ------------------------------------------------------------------
def _get_ip_from_ioctl(self) -> Optional[str]:
"""Get IPv4 address using ioctl SIOCGIFADDR."""
if not self.iface:
return None
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ifreq = struct.pack('16s256s', self.iface.encode(), b'\x00'*256)
res = fcntl.ioctl(sock.fileno(), 0x8915, ifreq) # SIOCGIFADDR
sock.close()
ip_bytes = res[20:24]
return socket.inet_ntoa(ip_bytes)
except Exception:
return self._get_ip_from_sys()
def _get_ip_from_sys(self) -> Optional[str]:
"""Fallback: read IP from /sys/class/net/<iface>/address."""
if not self.iface:
return None
path = f'/sys/class/net/{self.iface}/address'
# Note: /sys doesn't directly provide IPv4, try /proc instead
return self._get_ip_from_proc()
def _get_ip_from_proc(self) -> Optional[str]:
"""Read IPv4 from /proc/net/fib_trie or /proc/net/route."""
if not self.iface:
return None
try:
with open('/proc/net/fib_trie', 'r') as f:
content = f.read()
# Parse the trie for the interface
lines = content.split('\n')
in_correct_host = False
for line in lines:
if f"|-- {self.iface}" in line:
in_correct_host = True
elif in_correct_host and '|--' in line and 'host' in line:
parts = line.split()
for part in parts:
if re.match(r'^\d+\.\d+\.\d+\.\d+$', part):
return part
in_correct_host = False
except Exception:
pass
return None
def _get_netmask_from_ioctl(self) -> Optional[str]:
"""Get netmask using ioctl SIOCGIFNETMASK."""
if not self.iface:
return None
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ifreq = struct.pack('16s256s', self.iface.encode(), b'\x00'*256)
res = fcntl.ioctl(sock.fileno(), 0x891b, ifreq) # SIOCGIFNETMASK
sock.close()
mask_bytes = res[20:24]
return socket.inet_ntoa(mask_bytes)
except Exception:
return None
def _get_mac_from_sys(self) -> Optional[str]:
"""Read MAC address from /sys/class/net/<iface>/address."""
if not self.iface:
return None
path = f'/sys/class/net/{self.iface}/address'
try:
with open(path, 'r') as f:
mac = f.read().strip()
if re.match(r'([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}', mac):
return mac.upper()
except Exception:
pass
return None
def _get_stats_from_sys(self) -> Tuple[Optional[int], Optional[int]]:
"""Read rx_bytes and tx_bytes from /sys/class/net/<iface>/statistics/."""
if not self.iface:
return None, None
rx_path = f'/sys/class/net/{self.iface}/statistics/rx_bytes'
tx_path = f'/sys/class/net/{self.iface}/statistics/tx_bytes'
rx = self._read_int_file(rx_path)
tx = self._read_int_file(tx_path)
return rx, tx
def _get_gateway_from_proc(self) -> Optional[str]:
"""Parse /proc/net/route to find default gateway for the interface."""
if not self.iface:
return None
try:
with open('/proc/net/route', 'r') as f:
lines = f.readlines()[1:] # skip header
for line in lines:
parts = line.strip().split()
if len(parts) >= 3 and parts[0] == self.iface:
dest = int(parts[1], 16)
gateway = int(parts[2], 16)
if dest == 0: # default route
return self._int_to_ip(gateway)
except Exception:
pass
return None
def _get_ipv6_from_proc(self) -> Optional[List[str]]:
"""Parse /proc/net/if_inet6 for IPv6 addresses of the interface."""
if not self.iface:
return None
ipv6_list = []
try:
with open('/proc/net/if_inet6', 'r') as f:
for line in f:
parts = line.strip().split()
if len(parts) >= 4 and parts[3] == self.iface:
ipv6_hex = parts[0]
# Convert hex to IPv6 notation
ipv6 = ':'.join(ipv6_hex[i:i+4] for i in range(0, 32, 4))
# Compress IPv6 (optional)
ipv6 = self._compress_ipv6(ipv6)
ipv6_list.append(ipv6)
return ipv6_list if ipv6_list else None
except Exception:
return None
@staticmethod
def _compress_ipv6(ipv6: str) -> str:
"""Compress IPv6 address."""
try:
# Expand to full 8 groups
groups = ipv6.split(':')
if len(groups) < 8:
groups = ['0'] * (8 - len(groups)) + groups
# Find longest run of zeros
best_start = -1
best_len = 0
current_start = -1
current_len = 0
for i, group in enumerate(groups):
if group == '0':
if current_start == -1:
current_start = i
current_len += 1
else:
if current_len > best_len:
best_start = current_start
best_len = current_len
current_start = -1
current_len = 0
if current_len > best_len:
best_start = current_start
best_len = current_len
if best_len > 1:
# Compress
start = best_start
end = best_start + best_len
compressed = groups[:start] + [''] + groups[end:]
ipv6 = ':'.join(compressed).replace(':::', '::')
else:
# Remove leading zeros from each group
ipv6 = ':'.join(g.lstrip('0') or '0' for g in groups)
return ipv6
except Exception:
return ipv6
@staticmethod
def _read_int_file(path: str) -> Optional[int]:
"""Read integer from a file."""
try:
with open(path, 'r') as f:
return int(f.read().strip())
except Exception:
return None
@staticmethod
def _ip_to_int(ip: str) -> int:
"""Convert dotted IPv4 to integer."""
parts = ip.split('.')
return (int(parts[0]) << 24) + (int(parts[1]) << 16) + (int(parts[2]) << 8) + int(parts[3])
@staticmethod
def _int_to_ip(ip_int: int) -> str:
"""Convert integer to dotted IPv4."""
return f"{(ip_int >> 24) & 0xFF}.{(ip_int >> 16) & 0xFF}.{(ip_int >> 8) & 0xFF}.{ip_int & 0xFF}"
@staticmethod
def _bytes_to_human(bytes_val: Optional[int]) -> Optional[str]:
"""Convert bytes to human-readable format."""
if bytes_val is None:
return None
for unit in ['B', 'KiB', 'MiB', 'GiB', 'TiB']:
if bytes_val < 1024.0:
return f"{bytes_val:.2f} {unit}"
bytes_val /= 1024.0
return f"{bytes_val:.2f} PiB"
# ------------------------------------------------------------------
# Utility methods
# ------------------------------------------------------------------
def reset_to_system(self) -> 'Network':
"""Reset manually set values back to system values."""
self._manual_ip = False
self._manual_netmask = False
self._refresh_all()
return self
def __repr__(self) -> str:
if not self.iface:
return "Network(no interface set)"
manual_info = []
if self._manual_ip:
manual_info.append("manual_ip")
if self._manual_netmask:
manual_info.append("manual_netmask")
manual_str = f", manual={manual_info}" if manual_info else ""
return (f"Network(iface={self.iface}, ip={self._ip}, netmask={self._netmask}, "
f"gateway={self._gateway}, mac={self._mac}, cidr={self._cidr}, "
f"ipv6={self._ipv6}{manual_str})")
# ------------------------------------------------------------------
# Example usage and test
# ------------------------------------------------------------------
if __name__ == '__main__':
net = Network()
net.setiface('eth0') # Change to your interface (e.g., 'wlan0', 'enp0s3')
print("=== System Interface Information ===")
print(f"Interface: {net.iface}")
print(f"IP: {net.getip()}")
print(f"Netmask: {net.getnetmask()}")
print(f"Network: {net.getnetwork()}")
print(f"Gateway: {net.getgateway()}")
print(f"MAC: {net.getmac()}")
print(f"CIDR: {net.getcidr()}")
print(f"IPv6: {net.getipv6()}")
rx, tx = net.get_transfer_human()
print(f"RX bytes: {rx}")
print(f"TX bytes: {tx}")
print("\n=== Testing Custom IP/Netmask ===")
# Set custom IP and netmask
net.setip('10.1.1.1')
net.setnetmask('255.255.255.0')
# Or use CIDR notation: net.setnetmask('24')
print(f"Current IP: {net.getip()} (manually set)")
print(f"Current Netmask: {net.getnetmask()} (manually set)")
print(f"Network: {net.getnetwork()}")
print(f"CIDR: {net.getcidr()}")
print("\n=== IP Iteration Tests ===")
# Test with current settings
print(f"Using IP {net.getip()}, Netmask {net.getnetmask()}:")
print(f"inc=1 -> {net.iter_ip(1)}")
print(f"inc=2 -> {net.iter_ip(2)}")
print(f"inc=-1 -> {net.iter_ip(-1)}")
print(f"inc=-2 -> {net.iter_ip(-2)}")
# Test with custom values directly in iter_ip
print("\n=== Custom Values in iter_ip ===")
print(f"iter_ip(1, '192.168.1.1', '24'): {net.iter_ip(1, '192.168.1.1', '24')}")
print(f"iter_ip(-1, '192.168.1.1', '255.255.255.0'): {net.iter_ip(-1, '192.168.1.1', '255.255.255.0')}")
print(f"iter_ip(5, '10.0.0.1', '255.255.0.0'): {net.iter_ip(5, '10.0.0.1', '255.255.0.0')}")
# Generate a range of IPs
print("\n=== IP Range Generation ===")
range_ips = net.iter_ip_range('10.1.1.1', '24', 5)
print(f"Next 5 IPs: {range_ips}")
# Reset back to system values
net.reset_to_system()
print(f"\nAfter reset: IP={net.getip()} (back to system value)")

922
lib/sunhpc/sunhpc/output.py Normal file
View File

@@ -0,0 +1,922 @@
"""
Output class for creating beautifully aligned tables with flexible data input.
Supports headers, multiple data formats (list, tuple, dict), automatic column sizing,
and customizable empty value handling.
"""
from decimal import Decimal
from sunhpc.logs import Logs
from typing import List, Dict, Any, Optional, Union, Tuple
slog = Logs(name="Output", show_time=True)
class Output:
"""
A flexible table class for aligned output with header support.
Examples:
table = Output()
table.add_header(['name', 'age', 'city'])
table.add_row(['Alice', 28, 'New York'])
table.add_row({'name': 'Bob', 'age': 32, 'city': 'Los Angeles'})
table.add_row(('Charlie', 25, 'Chicago'))
table.print_table()
"""
def __init__(
self,
headers: Optional[List[str]] = None,
align: str = 'left',
empty_fill: str = 'N/A',
separator: str = ' ',
border_style: str = 'simple', # 'simple', 'grid', 'none'
header_separator_char: str = '-',
max_width: Optional[int] = None,
truncate_suffix: str = '...',
auto_align_numeric: bool = True
):
"""
Initialize the Output.
Args:
headers: Initial list of column headers
align: Default alignment for columns ('left', 'right', 'center')
empty_fill: String to fill empty/missing values
separator: Column separator string
border_style: Style of borders ('simple', 'grid', 'none')
header_separator_char: Character for header underline
max_width: Maximum width for any column (truncates if exceeded)
truncate_suffix: Suffix for truncated values
auto_align_numeric: Automatically right-align numeric values
"""
self.headers: List[str] = headers if headers else []
self.data: List[List[Any]] = []
self.alignments: Dict[str, str] = {}
self.default_align = align
self.empty_fill = empty_fill
self.separator = separator
self.border_style = border_style
self.header_separator_char = header_separator_char
self.max_width = max_width
self.truncate_suffix = truncate_suffix
self.auto_align_numeric = auto_align_numeric
# Column order (if headers are provided)
self.column_order = headers.copy() if headers else []
# Initialize alignments for headers
if headers:
for header in headers:
self.alignments[header] = align
def add_header(self, headers: List[str]) -> 'Output':
"""
Add or replace table headers.
Args:
headers: List of column header names
Returns:
Self for method chaining
"""
self.headers = headers
self.column_order = headers.copy()
# Initialize alignments for new headers
for header in headers:
if header not in self.alignments:
self.alignments[header] = self.default_align
return self
def set_alignment(self, column: str, align: str) -> 'Output':
"""
Set alignment for a specific column.
Args:
column: Column header name
align: Alignment ('left', 'right', 'center')
Returns:
Self for method chaining
"""
if column in self.headers:
self.alignments[column] = align
return self
def set_empty_fill(self, empty_fill: str) -> 'Output':
"""
Set the string to fill empty/missing values.
Args:
empty_fill: String to use for filling empty/missing values
Returns:
Self for method chaining
"""
self.empty_fill = empty_fill
return self
def set_separator(self, separator: str) -> 'Output':
"""
Set the column separator string.
Args:
separator: String to use for separating columns
Returns:
Self for method chaining
"""
self.separator = separator
return self
def set_border_style(self, border_style: str) -> 'Output':
"""
Set the style of borders.
Args:
border_style: Style of borders ('simple', 'grid', 'none')
Returns:
Self for method chaining
"""
self.border_style = border_style
return self
def set_header_separator_char(self, header_separator_char: str) -> 'Output':
"""
Set the character for header underline.
Args:
header_separator_char: Character to use for header underline
Returns:
Self for method chaining
"""
self.header_separator_char = header_separator_char
return self
def set_max_width(self, max_width: Optional[int] = None) -> 'Output':
"""
Set the maximum width for any column.
Args:
max_width: Maximum width for any column (truncates if exceeded)
Returns:
Self for method chaining
"""
self.max_width = max_width
return self
def set_truncate_suffix(self, truncate_suffix: str = '...') -> 'Output':
"""
Set the suffix for truncated values.
Args:
truncate_suffix: Suffix to use for truncated values
Returns:
Self for method chaining
"""
self.truncate_suffix = truncate_suffix
return self
def set_auto_align_numeric(self, auto_align_numeric: bool = True) -> 'Output':
"""
Set whether to automatically align numeric values.
Args:
auto_align_numeric: Whether to align numeric values ('left' or 'right')
Returns:
Self for method chaining
"""
self.auto_align_numeric = auto_align_numeric
return self
def add_row(
self,
row: Union[List[Any], Tuple[Any, ...], Dict[str, Any]],
auto_fill: bool = True
) -> 'Output':
"""
Add a row to the table.
Args:
row: Row data as list, tuple, or dictionary
auto_fill: Automatically fill missing columns with empty_fill
Returns:
Self for method chaining
"""
if not self.headers:
raise ValueError("Headers must be set before adding rows. Use add_header() first.")
processed_row = []
if isinstance(row, dict):
# Dictionary format: map values by header
for header in self.headers:
value = row.get(header)
if value is None and header not in row:
if auto_fill:
value = self.empty_fill
else:
value = ''
processed_row.append(value)
elif isinstance(row, (list, tuple)):
# List/tuple format: match by position
row_list = list(row)
for i, header in enumerate(self.headers):
if i < len(row_list):
processed_row.append(row_list[i])
elif auto_fill:
processed_row.append(self.empty_fill)
else:
processed_row.append('')
else:
raise TypeError(f"Row must be list, tuple, or dict, got {type(row)}")
self.data.append(processed_row)
return self
def add_rows(
self,
rows: List[Union[List[Any], Tuple[Any, ...], Dict[str, Any]]],
auto_fill: bool = True
) -> 'Output':
"""
Add multiple rows to the table.
Args:
rows: List of rows (each row as list, tuple, or dict)
auto_fill: Automatically fill missing columns with empty_fill
Returns:
Self for method chaining
"""
for row in rows:
self.add_row(row, auto_fill)
return self
def insert_row(
self,
index: int,
row: Union[List[Any], Tuple[Any, ...], Dict[str, Any]],
auto_fill: bool = True
) -> 'Output':
"""
Insert a row at a specific index.
Args:
index: Position to insert the row
row: Row data as list, tuple, or dictionary
auto_fill: Automatically fill missing columns with empty_fill
Returns:
Self for method chaining
"""
if not self.headers:
raise ValueError("Headers must be set before inserting rows. Use add_header() first.")
processed_row = []
if isinstance(row, dict):
for header in self.headers:
value = row.get(header)
if value is None and header not in row:
if auto_fill:
value = self.empty_fill
else:
value = ''
processed_row.append(value)
elif isinstance(row, (list, tuple)):
row_list = list(row)
for i, header in enumerate(self.headers):
if i < len(row_list):
processed_row.append(row_list[i])
elif auto_fill:
processed_row.append(self.empty_fill)
else:
processed_row.append('')
else:
raise TypeError(f"Row must be list, tuple, or dict, got {type(row)}")
self.data.insert(index, processed_row)
return self
def remove_row(self, index: int) -> 'Output':
"""
Remove a row at a specific index.
Args:
index: Position of row to remove
Returns:
Self for method chaining
"""
if 0 <= index < len(self.data):
self.data.pop(index)
return self
def clear(self) -> 'Output':
"""Clear all data rows but keep headers."""
self.data = []
return self
def reset(self) -> 'Output':
"""Reset the entire table (headers and data)."""
self.headers = []
self.data = []
self.column_order = []
self.alignments = {}
return self
def _calculate_column_widths(self) -> Dict[str, int]:
"""Calculate optimal width for each column."""
if not self.headers:
return {}
widths = {}
# Initialize with header widths
for i, header in enumerate(self.headers):
widths[header] = len(str(header))
# Check data widths
for row in self.data:
for i, header in enumerate(self.headers):
if i < len(row):
value = row[i]
# Handle None values
if value is None:
value = ''
# Handle special types
if isinstance(value, (int, float, Decimal)):
# Format numbers for width calculation
if isinstance(value, float):
value_str = f"{value:.2f}" if value == int(value) else str(value)
else:
value_str = str(value)
else:
value_str = str(value)
# Apply max width truncation for calculation
if self.max_width and len(value_str) > self.max_width:
value_str = value_str[:self.max_width - len(self.truncate_suffix)] + self.truncate_suffix
widths[header] = max(widths[header], len(value_str))
return widths
def _format_value(self, value: Any, width: int, column: str) -> str:
"""Format a single value with proper alignment and truncation."""
# Handle None values
if value is None:
value = self.empty_fill
# Convert to string
if isinstance(value, (int, float, Decimal)):
# Auto-right align numbers if enabled
if self.auto_align_numeric and self.alignments.get(column, self.default_align) == 'left':
align = 'right'
else:
align = self.alignments.get(column, self.default_align)
# Format numbers nicely
if isinstance(value, float):
# Show 2 decimal places for floats that are not integers
if value == int(value):
value_str = str(int(value))
else:
value_str = f"{value:.2f}"
else:
value_str = str(value)
elif isinstance(value, bool):
value_str = "True" if value else "False"
align = self.alignments.get(column, self.default_align)
else:
value_str = str(value)
align = self.alignments.get(column, self.default_align)
# Truncate if exceeds max width
if self.max_width and len(value_str) > self.max_width:
value_str = value_str[:self.max_width - len(self.truncate_suffix)] + self.truncate_suffix
# Apply alignment
if align == 'left':
return value_str.ljust(width)
elif align == 'right':
return value_str.rjust(width)
elif align == 'center':
return value_str.center(width)
else:
return value_str.ljust(width)
def _format_row(self, row: List[Any], widths: Dict[str, int]) -> List[str]:
"""Format an entire row."""
formatted_cells = []
for i, header in enumerate(self.headers):
if i < len(row):
value = row[i]
else:
value = self.empty_fill
formatted_cells.append(self._format_value(value, widths[header], header))
return formatted_cells
def _get_border_line(self, widths: Dict[str, int], style: str = 'simple') -> str:
"""Generate border line based on style."""
if style == 'none':
return ''
parts = []
for i, header in enumerate(self.headers):
width = widths[header]
if style == 'grid':
parts.append('+' + '-' * (width + 2))
else: # simple
parts.append('-' * (width + 2))
if style == 'grid':
return '+' + '+'.join(parts) + '+'
else:
return ' ' + ' '.join(parts)
def _format_header(self, widths: Dict[str, int]) -> str:
"""Format the header row."""
header_cells = []
for i, header in enumerate(self.headers):
align = self.alignments.get(header, self.default_align)
width = widths[header]
if align == 'left':
formatted = header.ljust(width)
elif align == 'right':
formatted = header.rjust(width)
else:
formatted = header.center(width)
header_cells.append(formatted)
if self.border_style == 'grid':
return '| ' + f' {self.separator} '.join(header_cells) + ' |'
else:
return self.separator.join(header_cells)
def _format_separator(self, widths: Dict[str, int]) -> str:
"""Format the separator line under headers."""
if self.border_style == 'none':
return ''
parts = []
for i, header in enumerate(self.headers):
width = widths[header]
parts.append(self.header_separator_char * width)
if self.border_style == 'grid':
return '|-' + f'-{self.separator}-'.join(parts) + '-|'
else:
return self.separator.join(parts)
def _format_data_row(self, row: List[str], is_last: bool = False) -> str:
"""Format a data row."""
if self.border_style == 'grid':
return '| ' + f' {self.separator} '.join(row) + ' |'
else:
return self.separator.join(row)
def get_table_string(self) -> str:
"""
Generate the complete table as a string.
Returns:
Formatted table string
"""
if not self.headers:
return "No headers defined"
if not self.data:
return f"Table '{self.separator.join(self.headers)}' has no data"
# Calculate column widths
widths = self._calculate_column_widths()
# Build table lines
lines = []
# Header
header_line = self._format_header(widths)
lines.append(header_line)
# Separator
separator_line = self._format_separator(widths)
if separator_line:
lines.append(separator_line)
# Data rows
for i, row in enumerate(self.data):
formatted_row = self._format_row(row, widths)
lines.append(self._format_data_row(formatted_row, i == len(self.data) - 1))
return '\n'.join(lines)
def print_table(self) -> None:
"""Print the table to console."""
print(self.get_table_string())
def export_to_csv(self, filename: str, include_header: bool = True) -> None:
"""
Export table to CSV file.
Args:
filename: Output file path
include_header: Whether to include header row
"""
import csv
with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
if include_header:
writer.writerow(self.headers)
for row in self.data:
# Convert values to strings and handle None
processed_row = [str(cell) if cell is not None else '' for cell in row]
writer.writerow(processed_row)
def get_data_frame(self) -> List[Dict[str, Any]]:
"""
Convert table data to list of dictionaries.
Returns:
List of dictionaries representing the table data
"""
result = []
for row in self.data:
row_dict = {}
for i, header in enumerate(self.headers):
if i < len(row):
value = row[i]
# Convert None to empty_fill for consistency
if value is None:
value = self.empty_fill
row_dict[header] = value
else:
row_dict[header] = self.empty_fill
result.append(row_dict)
return result
def sort_by(self, column: str, reverse: bool = False) -> 'Output':
"""
Sort table rows by a specific column.
Args:
column: Column header to sort by
reverse: Reverse sort order
Returns:
Self for method chaining
"""
if column not in self.headers:
raise ValueError(f"Column '{column}' not found in headers")
column_index = self.headers.index(column)
# Sort the data
self.data.sort(
key=lambda row: row[column_index] if column_index < len(row) and row[column_index] is not None else '',
reverse=reverse
)
return self
def filter_by(self, column: str, predicate) -> 'Output':
"""
Filter rows based on a predicate function.
Args:
column: Column header to filter by
predicate: Function that takes a value and returns bool
Returns:
Self for method chaining
"""
if column not in self.headers:
raise ValueError(f"Column '{column}' not found in headers")
column_index = self.headers.index(column)
# Filter the data
self.data = [
row for row in self.data
if column_index < len(row) and predicate(row[column_index])
]
return self
def column_exists(self, column: str) -> bool:
"""Check if a column exists."""
return column in self.headers
def get_column_index(self, column: str) -> Optional[int]:
"""Get the index of a column."""
if column in self.headers:
return self.headers.index(column)
return None
def get_column_data(self, column: str) -> List[Any]:
"""
Get all values from a specific column.
Args:
column: Column header name
Returns:
List of values from that column
"""
if column not in self.headers:
raise ValueError(f"Column '{column}' not found")
column_index = self.headers.index(column)
return [row[column_index] if column_index < len(row) else self.empty_fill for row in self.data]
def update_column(self, column: str, values: List[Any]) -> 'Output':
"""
Update values in a specific column.
Args:
column: Column header name
values: New values for the column (must match row count)
Returns:
Self for method chaining
"""
if column not in self.headers:
raise ValueError(f"Column '{column}' not found")
if len(values) != len(self.data):
raise ValueError(f"Number of values ({len(values)}) does not match number of rows ({len(self.data)})")
column_index = self.headers.index(column)
for i, value in enumerate(values):
if i < len(self.data):
if column_index >= len(self.data[i]):
# Extend the row if needed
self.data[i].extend([self.empty_fill] * (column_index - len(self.data[i]) + 1))
self.data[i][column_index] = value
return self
def get_statistics(self) -> Dict[str, Any]:
"""
Calculate statistics for numeric columns.
Returns:
Dictionary with statistics for each numeric column
"""
stats = {}
for column in self.headers:
column_data = self.get_column_data(column)
numeric_data = []
for value in column_data:
try:
if isinstance(value, (int, float)):
numeric_data.append(float(value))
elif isinstance(value, str) and value.replace('.', '').replace('-', '').isdigit():
numeric_data.append(float(value))
except (ValueError, TypeError):
pass
if numeric_data:
stats[column] = {
'count': len(numeric_data),
'min': min(numeric_data),
'max': max(numeric_data),
'sum': sum(numeric_data),
'avg': sum(numeric_data) / len(numeric_data),
'total_rows': len(column_data)
}
return stats
def dict_output(self, data, key_width=None, value_width=None, separator=':', indent=0):
'''
Print a dictionary as a table.
Args:
data: Dictionary to print
key_width: Width of the key column
value_width: Width of the value column
separator: Separator between key and value
indent: Indentation level
Returns:
Self for method chaining
'''
if not isinstance(data, dict):
try:
if isinstance(data, (list, tuple)) and len(data) % 2 == 0:
data = dict(zip(data[::2], data[1::2]))
else:
slog.error(f"Error: Cannot convert {type(data).__name__} to dict")
return self
except:
slog.error(f"Error: Cannot convert {type(data).__name__} to dict")
return self
if not data: return
# 自动计算键和值的宽度最大宽度,确保对齐
if key_width is None:
key_width = max(len(key) for key in data.keys())
if value_width is None:
value_width = max(len(str(value)) for value in data.values())
# 缩进
indent_str = ' ' * indent
# 输出每一行
for key, value in data.items():
print(f"{indent_str}{str(key):<{key_width}}{separator}{str(value):>{value_width}}")
return self
def __len__(self) -> int:
"""Return number of rows."""
return len(self.data)
def __repr__(self) -> str:
"""String representation."""
if not self.headers:
return "Output(no headers)"
return f"Output(headers={self.headers}, rows={len(self.data)})"
# ------------------------------------------------------------------
# Example usage and test
# ------------------------------------------------------------------
if __name__ == '__main__':
print("=" * 80)
print("ALIGNED TABLE DEMONSTRATION")
print("=" * 80)
# Example 1: Basic usage with list rows
print("\n1. Basic table with list rows:")
table1 = Output()
table1.add_header(['Name', 'Age', 'City', 'Country'])
table1.add_row(['Alice', 28, 'New York', 'USA'])
table1.add_row(['Bob', 32, 'Los Angeles', 'USA'])
table1.add_row(['Charlie', 25, 'Chicago', 'USA'])
table1.print_table()
# Example 2: Dictionary rows with missing values
print("\n2. Dictionary rows with missing values (auto-filled with '--'):")
table2 = Output(headers=['Name', 'Age', 'Email', 'Phone'])
table2.add_row({'Name': 'Alice', 'Age': 28, 'Email': 'alice@example.com'})
table2.add_row({'Name': 'Bob', 'Age': 32, 'Phone': '123-456-7890'})
table2.add_row({'Name': 'Charlie', 'Email': 'charlie@example.com', 'Phone': '098-765-4321'})
table2.print_table()
# Example 3: Mixed data types and custom alignment
print("\n3. Mixed data types with custom alignment:")
table3 = Output(headers=['Product', 'Price', 'Quantity', 'In Stock'])
table3.set_alignment('Price', 'right')
table3.set_alignment('Quantity', 'right')
table3.add_rows([
['Laptop', 999.99, 5, True],
['Mouse', 19.99, 25, True],
['Keyboard', 49.99, 0, False],
['Monitor', 299.99, 3, True]
])
table3.print_table()
# Example 4: Different border styles
print("\n4. Grid border style:")
table4 = Output(headers=['ID', 'Name', 'Score'], border_style='grid')
table4.add_rows([
[1, 'John', 85.5],
[2, 'Jane', 92.0],
[3, 'Bob', 78.5]
])
table4.print_table()
print("\n5. No border style:")
table4.border_style = 'none'
table4.print_table()
# Example 6: Tuple rows
print("\n6. Tuple rows:")
table6 = Output(headers=['X', 'Y', 'Z'])
table6.add_rows([
(10, 20, 30),
(40, 50, 60),
(70, 80, 90)
])
table6.print_table()
# Example 7: Sorting and filtering
print("\n7. Sorting by age (descending):")
table7 = Output(headers=['Name', 'Age', 'Salary'])
table7.add_rows([
{'Name': 'Alice', 'Age': 28, 'Salary': 50000},
{'Name': 'Bob', 'Age': 35, 'Salary': 60000},
{'Name': 'Charlie', 'Age': 25, 'Salary': 45000},
{'Name': 'Diana', 'Age': 30, 'Salary': 55000}
])
table7.sort_by('Age', reverse=True)
table7.print_table()
print("\n8. Filtered (salary > 50000):")
table8 = Output(headers=['Name', 'Age', 'Salary'])
table8.add_rows([
{'Name': 'Alice', 'Age': 28, 'Salary': 50000},
{'Name': 'Bob', 'Age': 35, 'Salary': 60000},
{'Name': 'Charlie', 'Age': 25, 'Salary': 45000},
{'Name': 'Diana', 'Age': 30, 'Salary': 55000}
])
table8.filter_by('Salary', lambda x: x > 50000 if x is not None else False)
table8.print_table()
# Example 9: Column operations
print("\n9. Column operations:")
table9 = Output(headers=['Product', 'Price', 'Quantity'])
table9.add_rows([
['Apple', 1.2, 100],
['Banana', 0.8, 150],
['Orange', 1.5, 80]
])
print("Column 'Price' values:", table9.get_column_data('Price'))
# Calculate total value
prices = table9.get_column_data('Price')
quantities = table9.get_column_data('Quantity')
totals = [p * q for p, q in zip(prices, quantities)]
table9.add_header(['Product', 'Price', 'Quantity', 'Total'])
table9.update_column('Total', totals)
table9.print_table()
# Example 10: Statistics
print("\n10. Statistics for numeric columns:")
table10 = Output(headers=['Name', 'Math', 'Science', 'English'])
table10.add_rows([
['Alice', 85, 90, 88],
['Bob', 78, 85, 82],
['Charlie', 92, 88, 91],
['Diana', 88, 92, 87]
])
stats = table10.get_statistics()
for column, stat in stats.items():
if column != 'Name':
print(f" {column}: avg={stat['avg']:.1f}, min={stat['min']}, max={stat['max']}")
# Example 11: Export to CSV (uncomment to test)
# table10.export_to_csv('grades.csv')
# Example 12: Complex real-world example
print("\n11. Real-world example - Employee Directory:")
employees = Output(
headers=['Employee ID', 'Name', 'Department', 'Position', 'Salary', 'Status'],
align='left',
empty_fill='N/A',
separator='',
border_style='grid'
)
employees.set_alignment('Salary', 'right')
employees.set_alignment('Employee ID', 'center')
employees.add_rows([
{'Employee ID': 1001, 'Name': 'John Smith', 'Department': 'Engineering', 'Position': 'Senior Developer', 'Salary': 95000, 'Status': 'Active'},
{'Employee ID': 1002, 'Name': 'Jane Doe', 'Department': 'Marketing', 'Position': 'Manager', 'Salary': 85000, 'Status': 'Active'},
{'Employee ID': 1003, 'Name': 'Bob Johnson', 'Department': 'Engineering', 'Position': 'Developer', 'Salary': 75000, 'Status': 'On Leave'},
{'Employee ID': 1004, 'Name': 'Alice Brown', 'Department': 'Sales', 'Position': 'Representative', 'Salary': 65000, 'Status': 'Active'},
{'Employee ID': 1005, 'Name': 'Charlie Wilson', 'Department': 'Engineering', 'Position': 'Lead', 'Salary': 110000, 'Status': 'Active'},
{'Employee ID': 1006, 'Name': 'Diana Miller', 'Department': 'HR', 'Position': 'Specialist', 'Salary': 60000, 'Status': 'Inactive'},
])
employees.print_table()
# Example 13: Get data as dictionary list
print("\n12. Export to dictionary list:")
data_dict = table10.get_data_frame()
for row in data_dict[:2]: # Show first 2 rows
print(f" {row}")
# Example 14: Chaining methods
print("\n13. Method chaining example:")
(Output(headers=['Item', 'Count'])
.add_row(['Apples', 10])
.add_row(['Bananas', 15])
.add_row(['Oranges', 8])
.sort_by('Count', reverse=True)
.print_table())

View File

@@ -1,8 +1,10 @@
import os import os
import sys import sys
import shutil
import xml.sax import xml.sax
import subprocess import subprocess
from xml.sax import handler from xml.sax import handler
from typing import Dict, Any, Optional, List, Union
class SunhpcException(Exception): class SunhpcException(Exception):
'Base class for Sunhpc exceptions.' 'Base class for Sunhpc exceptions.'
@@ -97,6 +99,15 @@ def getNativeArch():
return 'i386' return 'i386'
return arch return arch
def getTerminalWidth():
get_terminal_size = getattr(os, 'get_terminal_size', shutil.get_terminal_size)
try:
width = get_terminal_size().columns
except Exception:
width = 80
return width
def mkdir(newdir): def mkdir(newdir):
if os.path.isdir(newdir): if os.path.isdir(newdir):
pass pass
@@ -181,3 +192,4 @@ def startSpinner(cmd):
for i in range(0, 78): for i in range(0, 78):
pad = pad + ' ' pad = pad + ' '
print ('\r%s\r' % pad, end='', flush=True) print ('\r%s\r' % pad, end='', flush=True)