diff --git a/lib/sunhpc/sunhpc/commands/__init__.py b/lib/sunhpc/sunhpc/commands/__init__.py
index 3fbc549..f3d0897 100644
--- a/lib/sunhpc/sunhpc/commands/__init__.py
+++ b/lib/sunhpc/sunhpc/commands/__init__.py
@@ -3,9 +3,23 @@ import re
import sys
import pwd
import xml
+import yaml
+import types
import syslog
+import inspect
+import argparse
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 handler
from xml.sax import make_parser
@@ -312,21 +326,33 @@ class Command:
MustBeRoot = 1
def __init__(self, database):
- self.db = database
+ self.db = database
+ self.conf = Config('/etc/sunhpc/sunhpc.yaml')
+ self.text = ''
+ self.output = []
- self.text = ''
- self.output = []
+ self.net = Network()
+ self.mem = MemoryInfo()
+ self.disk = DiskInfo()
+ self.fmt = Output()
- self.os = os.uname()[0].lower()
- self.arch = os.uname()[4]
- self.user = pwd.getpwuid(os.getuid()).pw_name
+ self.os = os.uname()[0].lower()
+ self.arch = os.uname()[4]
+ self.user = pwd.getpwuid(os.getuid()).pw_name
- self._args = None
- self._params = None
+ self.parser = argparse.ArgumentParser()
+ self.subcommand = 'Sub Command'
+ self.subdesc = None
+ self.epilog = None
+
+ self._args = None
+ self._params = None
if 'SUNHPCDEBUG' in os.environ:
self._debug = True
+ self.log = Logs(name='Sunhpc', level=0, show_time=True)
else:
self._debug = False
+ self.log = Logs(name='Sunhpc', show_time=True)
def debug(self):
"""Return True if debug mode is enabled"""
@@ -366,6 +392,55 @@ class Command:
else:
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):
self.text = ''
def addText(self, s):
@@ -600,6 +675,9 @@ class Command:
return rlist
def runWrapper(self, name, args):
+
+ self.subcommand = name
+
if args:
command = '%s %s' % (name, ' '.join(args))
else:
@@ -689,4 +767,4 @@ class Command:
elif format == 'sphinx':
self.addText(handler.getSphinxText())
else:
- self.addText(handler.getPlainText())
\ No newline at end of file
+ self.addText(handler.getPlainText())
diff --git a/lib/sunhpc/sunhpc/commands/init/config/__init__.py b/lib/sunhpc/sunhpc/commands/init/config/__init__.py
index 7435706..10a8cff 100644
--- a/lib/sunhpc/sunhpc/commands/init/config/__init__.py
+++ b/lib/sunhpc/sunhpc/commands/init/config/__init__.py
@@ -3,6 +3,8 @@ import re
import sys
import sunhpc.commands
+from sunhpc.configs import Config
+
class Command(sunhpc.commands.init.command):
"""
这个命令是初始化系统的相关配置，并且写入到配置文件.
@@ -22,6 +24,12 @@ class Command(sunhpc.commands.init.command):
提供一个系统ISO文件、或者挂载路径、例如: /mnt/cdrom
+
+ -h --help 显示帮助信息
+ -d --debug 开启调试模式
+ -v --verbose 显示详细信息
+
+
使用eth1网络接口、和 /mnt/cdrom 挂载路径
@@ -30,6 +38,80 @@ class Command(sunhpc.commands.init.command):
使用eth1网络接口、和 /mnt/cdrom 挂载路径
"""
-
def run(self, params, args):
- print ('This is init opt ...')
\ No newline at end of file
+ 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)
+ '''
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/configs.py b/lib/sunhpc/sunhpc/configs.py
new file mode 100644
index 0000000..caa3d25
--- /dev/null
+++ b/lib/sunhpc/sunhpc/configs.py
@@ -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()
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/disks.py b/lib/sunhpc/sunhpc/disks.py
new file mode 100644
index 0000000..ca308e5
--- /dev/null
+++ b/lib/sunhpc/sunhpc/disks.py
@@ -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')}")
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/error.py b/lib/sunhpc/sunhpc/error.py
new file mode 100644
index 0000000..f6b603b
--- /dev/null
+++ b/lib/sunhpc/sunhpc/error.py
@@ -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
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/logs.py b/lib/sunhpc/sunhpc/logs.py
new file mode 100644
index 0000000..c35f89f
--- /dev/null
+++ b/lib/sunhpc/sunhpc/logs.py
@@ -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("=")
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/main.py b/lib/sunhpc/sunhpc/main.py
index bd9905a..4401415 100644
--- a/lib/sunhpc/sunhpc/main.py
+++ b/lib/sunhpc/sunhpc/main.py
@@ -6,9 +6,16 @@ import sunhpc
import signal
import syslog
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):
-
try:
database = None
except ImportError:
@@ -46,7 +53,7 @@ def _main(argv=None):
print ('error - invalid sunhpc command "%s"' % args[0])
sys.exit(-1)
- name = ' '.join('.'.split(s))
+ name = ' '.join(s.split('.')[2:])
try:
command = getattr(module, 'Command')(database)
@@ -63,36 +70,51 @@ def _main(argv=None):
command.isRootUser() or command.isApacheUser()):
os.system('sudo %s' % ' '.join(sys.argv))
else:
+ #import sunhpc
try:
command.runWrapper(name, args[i:])
text = command.getText()
if len(text) > 0:
- print (text, end=' ')
- if text[len(text)-1] != '\n':
- print ()
- except sunhpc.util.CommandError as e:
+ if text.endswith('\n'):
+ sys.stdout.write(text)
+ else:
+ print (text)
+ print ()
+ except CommandError as e:
print ('Error: ', e)
print (command.usage())
- exit(1)
-
- syslog.closelog()
+ 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()
+ return 0
def main(argv=None):
- _main(argv)
- '''
- try:
- return _main(argv)
- except KeyboardInterrupt:
- sys.stderr.write("\n")
- tty.error("Keyboard interrupt.")
- return signal.SIGINT.value
+ if any(flag in argv for flag in ['--debug', '-d']):
+ _main(argv)
+ else:
+ try:
+ return _main(argv)
+ except KeyboardInterrupt:
+ sys.stderr.write("\n")
+ tty.error("Keyboard interrupt.")
+ return signal.SIGINT.value
- except SystemExit as e:
- return e.code
+ except AttributeError:
+ sys.stderr.write("\n")
+ tty.error("Attribute error.")
+ return 1
- except Exception as e:
- tty.error(e)
- return 3
- '''
+ except SystemExit as e:
+ return e.code
+
+ except Exception as e:
+ tty.error(e)
+ return 3
diff --git a/lib/sunhpc/sunhpc/memory.py b/lib/sunhpc/sunhpc/memory.py
new file mode 100644
index 0000000..672cbdc
--- /dev/null
+++ b/lib/sunhpc/sunhpc/memory.py
@@ -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
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/network.py b/lib/sunhpc/sunhpc/network.py
new file mode 100644
index 0000000..7317279
--- /dev/null
+++ b/lib/sunhpc/sunhpc/network.py
@@ -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//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//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//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)")
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/output.py b/lib/sunhpc/sunhpc/output.py
new file mode 100644
index 0000000..f61a656
--- /dev/null
+++ b/lib/sunhpc/sunhpc/output.py
@@ -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())
\ No newline at end of file
diff --git a/lib/sunhpc/sunhpc/util.py b/lib/sunhpc/sunhpc/util.py
index 4a2551a..8e07793 100644
--- a/lib/sunhpc/sunhpc/util.py
+++ b/lib/sunhpc/sunhpc/util.py
@@ -1,8 +1,10 @@
import os
import sys
+import shutil
import xml.sax
import subprocess
from xml.sax import handler
+from typing import Dict, Any, Optional, List, Union
class SunhpcException(Exception):
'Base class for Sunhpc exceptions.'
@@ -97,6 +99,15 @@ def getNativeArch():
return 'i386'
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):
if os.path.isdir(newdir):
pass
@@ -181,3 +192,4 @@ def startSpinner(cmd):
for i in range(0, 78):
pad = pad + ' '
print ('\r%s\r' % pad, end='', flush=True)
+