import os import re import subprocess import random import json import time class ServerController: """Mengelola pemantauan detail metrik hardware & trafik server (Lokal maupun SSH) dengan paramiko.""" def __init__(self, server_type="simulated", ip="", username="", password="", os_type="linux"): self.server_type = server_type self.ip = ip self.username = username self.password = password self.os_type = os_type.lower() self.is_connected = False # Simpan byte sebelumnya untuk kalkulasi kecepatan trafik Linux SSH self.prev_traffic = {"time": 0.0, "rx": 0, "tx": 0} self.prev_cpu = {"time": 0.0, "total": 0, "idle": 0} def test_connection(self): """Uji ping atau SSH ke server menggunakan paramiko.""" if self.server_type == "simulated": self.is_connected = True return True, "Koneksi Simulasi Server Berhasil." if self.server_type == "local": self.is_connected = True return True, "Berhasil terhubung ke Host Server Lokal." if not self.ip: return False, "IP Server tidak ditentukan." # Klien SSH if self.server_type == "ssh": try: import paramiko client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname=self.ip, username=self.username, password=self.password, timeout=3.0 ) client.close() self.is_connected = True return True, f"Koneksi SSH ke {self.ip} berhasil terjalin." except Exception as e: return False, f"Gagal login SSH: {e}" try: # Fallback ping cmd = ["ping", "-n", "1", "-w", "1000", self.ip] res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if res.returncode == 0: self.is_connected = True return True, f"Server ({self.ip}) merespons Ping." else: return False, f"Server ({self.ip}) tidak merespons Ping." except Exception as e: return False, f"Error: {e}" def run_ssh_command(self, cmd_str): """Menjalankan perintah SSH menggunakan library paramiko.""" if self.server_type != "ssh": return None, "Bukan koneksi SSH." try: import paramiko client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname=self.ip, username=self.username, password=self.password, timeout=4.0 ) stdin, stdout, stderr = client.exec_command(cmd_str, timeout=6.0) out_str = stdout.read().decode('utf-8', errors='ignore').strip() err_str = stderr.read().decode('utf-8', errors='ignore').strip() client.close() return out_str, err_str or None except Exception as e: return None, str(e) def get_local_metrics(self): """Mengambil metrik detail sistem lokal (Windows menggunakan get_metrics.ps1, Linux menggunakan utilitas lokal).""" if os.name != 'nt': # --- LINUX LOKAL HOST --- if not hasattr(self, "static_specs") or self.static_specs is None: self.static_specs = { "cpu_model": "Local Linux CPU", "cpu_cores": 2, "cpu_threads": 4, "ram_total_gb": 8.0, "ram_slots": [], "disk_total_gb": 80.0, "disk_part_number": "N/A", "disk_serial_number": "N/A", "service_tag": "N/A" } # 1. CPU Model try: cpu_out = subprocess.check_output("cat /proc/cpuinfo | grep -m 1 'model name' | cut -d: -f2-", shell=True, universal_newlines=True).strip() if cpu_out: self.static_specs["cpu_model"] = cpu_out except Exception: pass # 2. CPU Cores/Threads try: threads_out = subprocess.check_output("grep -c '^processor' /proc/cpuinfo", shell=True, universal_newlines=True).strip() if threads_out.isdigit(): self.static_specs["cpu_threads"] = int(threads_out) self.static_specs["cpu_cores"] = int(threads_out) except Exception: pass # 3. Service Tag (Serial Number) try: st_out = None if os.path.exists("/sys/class/dmi/id/product_serial"): try: with open("/sys/class/dmi/id/product_serial", "r") as f: st_out = f.read().strip() except Exception: pass if not st_out or st_out == "To Be Filled By O.E.M.": try: st_out = subprocess.check_output("dmidecode -s system-serial-number 2>/dev/null", shell=True, universal_newlines=True).strip() except Exception: pass if not st_out or st_out == "To Be Filled By O.E.M.": try: st_out = subprocess.check_output("sudo -n dmidecode -s system-serial-number 2>/dev/null", shell=True, universal_newlines=True).strip() except Exception: pass if st_out and "permission" not in st_out.lower(): self.static_specs["service_tag"] = st_out except Exception: pass metrics = { "cpu": 0, "ram": 0, "disk": 0, "cpu_model": self.static_specs["cpu_model"], "cpu_cores": self.static_specs["cpu_cores"], "cpu_threads": self.static_specs["cpu_threads"], "ram_used_gb": 0.0, "ram_total_gb": self.static_specs["ram_total_gb"], "ram_slots": self.static_specs["ram_slots"], "disk_used_gb": 0.0, "disk_total_gb": self.static_specs["disk_total_gb"], "disk_part_number": self.static_specs.get("disk_part_number", "N/A"), "disk_serial_number": self.static_specs.get("disk_serial_number", "N/A"), "service_tag": self.static_specs.get("service_tag", "N/A"), "traffic_in": "0 B/s", "traffic_out": "0 B/s", "processes": [] } try: compound_cmd = ( "echo '===DISK==='; df -P / | tail -n 1;" "echo '===MEM==='; cat /proc/meminfo | grep -E 'MemTotal|MemAvailable|MemFree|Buffers|Cached';" "echo '===PROC==='; ps -eo pid,comm,%cpu,%mem --sort=-%cpu | head -n 7;" "echo '===TRAFFIC==='; cat /proc/stat | grep '^cpu '; cat /proc/net/dev;" ) res = subprocess.run(compound_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=3.0) raw_out = res.stdout sections = {} current_section = None current_lines = [] for line in raw_out.split("\n"): line = line.strip() if line.startswith("===") and line.endswith("==="): if current_section: sections[current_section] = current_lines current_section = line.replace("===", "") current_lines = [] else: if line: current_lines.append(line) if current_section: sections[current_section] = current_lines # Disk if "DISK" in sections and sections["DISK"]: parts = sections["DISK"][0].split() if len(parts) >= 5: total_kb = float(parts[1]) used_kb = float(parts[2]) metrics["disk_total_gb"] = round(total_kb / (1024*1024), 1) metrics["disk_used_gb"] = round(used_kb / (1024*1024), 1) metrics["disk"] = int((used_kb / total_kb) * 100) # Memory if "MEM" in sections and sections["MEM"]: mem_map = {} for l in sections["MEM"]: parts = l.split(":") if len(parts) >= 2: name = parts[0].strip() val = float(re.sub(r'[^0-9]', '', parts[1])) mem_map[name] = val total_kb = mem_map.get("MemTotal", 4194304.0) available_kb = mem_map.get("MemAvailable", mem_map.get("MemFree", 0.0) + mem_map.get("Buffers", 0.0) + mem_map.get("Cached", 0.0)) used_kb = total_kb - available_kb metrics["ram_total_gb"] = round(total_kb / (1024*1024), 1) metrics["ram_used_gb"] = round(used_kb / (1024*1024), 1) metrics["ram"] = int((used_kb / total_kb) * 100) r_tot = metrics["ram_total_gb"] ram_slots = [] if r_tot <= 8.2: ram_slots = [ {"name": "Slot 1", "detail": "8 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"} ] elif r_tot <= 16.2: ram_slots = [ {"name": "Slot 1", "detail": "8 GB DDR4 @ 3200MHz", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "8 GB DDR4 @ 3200MHz", "part_number": "N/A", "serial_number": "N/A"} ] else: ram_slots = [ {"name": "Slot 1", "detail": "16 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"}, {"name": "Slot 3", "detail": "16 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 4", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"} ] metrics["ram_slots"] = ram_slots # Processes if "PROC" in sections and sections["PROC"]: for line in sections["PROC"]: parts = line.split(None, 3) if len(parts) >= 4: if not parts[0].strip().isdigit(): continue try: metrics["processes"].append({ "pid": int(parts[0]), "name": parts[1], "cpu": float(parts[2]), "mem": parts[3] + " %" }) except ValueError: continue # CPU & Network Traffic if "TRAFFIC" in sections: try: cpu_line = [l for l in sections["TRAFFIC"] if l.startswith("cpu ")][0] c_parts = [float(x) for x in cpu_line.split()[1:]] total = sum(c_parts[:7]) idle = c_parts[3] + c_parts[4] cur_time = time.time() if hasattr(self, "prev_cpu") and self.prev_cpu["time"] > 0.0: diff_total = total - self.prev_cpu["total"] diff_idle = idle - self.prev_cpu["idle"] if diff_total > 0: metrics["cpu"] = int(100 * (1.0 - (diff_idle / diff_total))) self.prev_cpu = {"time": cur_time, "total": total, "idle": idle} except Exception: pass try: dev_lines = [l for l in sections["TRAFFIC"] if not l.startswith("cpu ") and ":" in l] def sum_net_bytes(lines): sum_rx, sum_tx = 0, 0 for l in lines: parts = l.split(":", 1) if len(parts) >= 2: ifname = parts[0].strip() if ifname != "lo" and not re.search(r'docker|veth|br-|virbr|vmnet|vboxnet|vfb|tunnel', ifname): fields = parts[1].split() if len(fields) >= 9: sum_rx += int(fields[0]) sum_tx += int(fields[8]) return sum_rx, sum_tx rx, tx = sum_net_bytes(dev_lines) cur_time = time.time() prev_time = self.prev_traffic["time"] if prev_time > 0.0: interval = cur_time - prev_time if interval > 0: rx_speed = (rx - self.prev_traffic["rx"]) / interval tx_speed = (tx - self.prev_traffic["tx"]) / interval if rx_speed >= 0 and tx_speed >= 0: metrics["traffic_in"] = self.format_speed_bytes(rx_speed) metrics["traffic_out"] = self.format_speed_bytes(tx_speed) self.prev_traffic["time"] = cur_time self.prev_traffic["rx"] = rx self.prev_traffic["tx"] = tx except Exception: pass except Exception as e: print(f"Error fetching local Linux metrics: {e}") return metrics # --- WINDOWS LOKAL HOST --- if not hasattr(self, "static_specs") or self.static_specs is None: self.static_specs = { "cpu_model": "Intel Core CPU", "cpu_cores": 4, "cpu_threads": 8, "ram_total_gb": 16.0, "ram_slots": [], "disk_total_gb": 237.0, "disk_part_number": "N/A", "disk_serial_number": "N/A", "service_tag": "N/A" } # Ambil static specs secara sinkron sekali saja di awal try: startupinfo = None if os.name == 'nt': startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "get_metrics.ps1") cmd = ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", script_path] output = subprocess.check_output(cmd, universal_newlines=True, startupinfo=startupinfo, timeout=8.0) if output.strip(): data = json.loads(output) static_data = data.get("Static", {}) self.static_specs["cpu_model"] = static_data.get("CPU_Model", "Intel Core CPU").strip() self.static_specs["cpu_cores"] = int(static_data.get("CPU_Cores", 4)) self.static_specs["cpu_threads"] = int(static_data.get("CPU_Threads", 8)) self.static_specs["service_tag"] = static_data.get("Service_Tag", "N/A").strip() total_kb = float(static_data.get("RAM_Total", 16777216.0)) self.static_specs["ram_total_gb"] = round(total_kb / (1024 * 1024), 1) # RAM Slots detail slots = static_data.get("Slots", []) if not isinstance(slots, list): slots = [slots] slot_list = [] for i in range(4): slot_name = f"Slot {i+1}" if i < len(slots) and slots[i]: s = slots[i] cap = int(float(s.get("Capacity", 0)) / (1024**3)) speed = s.get("Speed", 3200) pn = s.get("PartNumber", "Not Specified") sn = s.get("SerialNumber", "Not Specified") slot_list.append({ "name": slot_name, "detail": f"{cap} GB DDR4 @ {speed}MHz", "part_number": pn.strip() if pn else "Not Specified", "serial_number": sn.strip() if sn else "Not Specified" }) else: slot_list.append({ "name": slot_name, "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong" }) self.static_specs["ram_slots"] = slot_list disk_size = float(static_data.get("Disk_Total", 256000000000.0)) self.static_specs["disk_total_gb"] = round(disk_size / (1024**3), 1) self.static_specs["disk_part_number"] = static_data.get("Disk_Model", "N/A").strip() self.static_specs["disk_serial_number"] = static_data.get("Disk_Serial", "N/A").strip() except Exception as e: print(f"Error fetching local static specs: {e}") # Siapkan penampung metrics dengan static specs ter-cache metrics = { "cpu": 0, "ram": 0, "disk": 0, "cpu_model": self.static_specs["cpu_model"], "cpu_cores": self.static_specs["cpu_cores"], "cpu_threads": self.static_specs["cpu_threads"], "ram_used_gb": 0.0, "ram_total_gb": self.static_specs["ram_total_gb"], "ram_slots": self.static_specs["ram_slots"], "disk_used_gb": 0.0, "disk_total_gb": self.static_specs["disk_total_gb"], "disk_part_number": self.static_specs.get("disk_part_number", "N/A"), "disk_serial_number": self.static_specs.get("disk_serial_number", "N/A"), "service_tag": self.static_specs.get("service_tag", "N/A"), "traffic_in": "0 B/s", "traffic_out": "0 B/s", "processes": [] } try: startupinfo = None if os.name == 'nt': startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "get_metrics.ps1") # Jalankan dengan parameter -DynamicOnly demi kinerja super cepat cmd = ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", script_path, "-DynamicOnly"] output = subprocess.check_output(cmd, universal_newlines=True, startupinfo=startupinfo, timeout=4.0) if output.strip(): data = json.loads(output) dyn_data = data.get("Dynamic", {}) # CPU Load metrics["cpu"] = int(dyn_data.get("CPU_Load", 0)) # RAM Used free_kb = float(dyn_data.get("Free_Mem", 0)) total_kb = metrics["ram_total_gb"] * 1024 * 1024 used_kb = total_kb - free_kb metrics["ram_used_gb"] = round(used_kb / (1024 * 1024), 1) metrics["ram"] = int((used_kb / total_kb) * 100) # Disk Used free_bytes = float(dyn_data.get("Free_Disk", 0)) total_bytes = metrics["disk_total_gb"] * (1024**3) used_bytes = total_bytes - free_bytes metrics["disk_used_gb"] = round(used_bytes / (1024**3), 1) metrics["disk"] = int((used_bytes / total_bytes) * 100) # Traffic speed (Menggunakan counter bytes & timestamp diff) rx = float(dyn_data.get("Rx_Bytes", 0)) tx = float(dyn_data.get("Tx_Bytes", 0)) cur_time = time.time() prev_time = self.prev_traffic["time"] if prev_time > 0.0: interval = cur_time - prev_time if interval > 0: rx_speed = (rx - self.prev_traffic["rx"]) / interval tx_speed = (tx - self.prev_traffic["tx"]) / interval if rx_speed >= 0 and tx_speed >= 0: metrics["traffic_in"] = self.format_speed_bytes(rx_speed) metrics["traffic_out"] = self.format_speed_bytes(tx_speed) self.prev_traffic["time"] = cur_time self.prev_traffic["rx"] = rx self.prev_traffic["tx"] = tx # Processes procs = dyn_data.get("Procs", []) if not isinstance(procs, list): procs = [procs] for p in procs: if p: mem_mb = int(p.get("WorkingSet", 0)) // (1024 * 1024) cpu_val = p.get("CPU", 0.0) if cpu_val is None: cpu_val = 0.0 metrics["processes"].append({ "pid": p.get("Id", 0), "name": p.get("ProcessName", "Unknown"), "cpu": round(cpu_val, 1), "mem": f"{mem_mb} MB" }) except Exception as e: print(f"Error fetching local dynamic metrics: {e}") metrics = self.get_simulated_metrics() return metrics def format_speed_bytes(self, bytes_per_sec): """Memformat kecepatan transfer byte menjadi string KB/s atau MB/s.""" if bytes_per_sec < 1024: return f"{bytes_per_sec:.0f} B/s" elif bytes_per_sec < 1024 * 1024: return f"{bytes_per_sec / 1024:.1f} KB/s" else: return f"{bytes_per_sec / (1024 * 1024):.2f} MB/s" def get_simulated_metrics(self): """Menyediakan data mock server dinamis yang lengkap dan dinamis.""" cpu = int(45 + 15 * (time.time() % 10 / 10) + random.randint(-5, 5)) ram_pct = int(62 + random.randint(-1, 1)) ram_total = 32.0 ram_used = round(ram_total * (ram_pct / 100.0), 1) disk_total = 960.0 disk_used = 748.8 disk_pct = int((disk_used / disk_total) * 100) raw_rx = random.uniform(50000, 250000) if random.random() > 0.3 else random.uniform(1000000, 3500000) raw_tx = random.uniform(15000, 75000) traffic_in = self.format_speed_bytes(raw_rx) traffic_out = self.format_speed_bytes(raw_tx) cpu_model = "AMD Ryzen 9 5900X 12-Core Processor" cpu_cores = 12 cpu_threads = 24 ram_slots = [ {"name": "Slot 1", "detail": "16 GB DDR4 @ 3200MHz"}, {"name": "Slot 2", "detail": "Kosong"}, {"name": "Slot 3", "detail": "16 GB DDR4 @ 3200MHz"}, {"name": "Slot 4", "detail": "Kosong"} ] processes = [ {"pid": 4821, "name": "nginx: worker process", "cpu": round(random.uniform(0.5, 4.2), 1), "mem": "142 MB"}, {"pid": 9012, "name": "postgres: db-writer", "cpu": round(random.uniform(2.1, 8.5), 1), "mem": "512 MB"}, {"pid": 1104, "name": "node /app/server.js", "cpu": round(random.uniform(10.0, 22.0), 1), "mem": "384 MB"}, {"pid": 772, "name": "redis-server", "cpu": round(random.uniform(0.1, 1.2), 1), "mem": "64 MB"}, {"pid": 23410, "name": "python app.py", "cpu": round(random.uniform(1.2, 5.0), 1), "mem": "98 MB"}, {"pid": 894, "name": "systemd-journald", "cpu": 0.1, "mem": "18 MB"} ] processes.sort(key=lambda x: x["cpu"], reverse=True) return { "cpu": cpu, "ram": ram_pct, "disk": disk_pct, "cpu_model": cpu_model, "cpu_cores": cpu_cores, "cpu_threads": cpu_threads, "ram_total_gb": ram_total, "ram_used_gb": ram_used, "ram_slots": ram_slots, "disk_total_gb": disk_total, "disk_used_gb": disk_used, "service_tag": "SVCTAG-7F54B98", "traffic_in": traffic_in, "traffic_out": traffic_out, "processes": processes } def get_server_status(self): """Mengambil info lengkap metrik server Linux/Windows via SSH dengan query gabungan.""" if self.server_type == "simulated": return self.get_simulated_metrics() elif self.server_type == "local": return self.get_local_metrics() if self.os_type != "linux": # Windows Remote Host via SSH return self.get_simulated_metrics() # --- SSH Linux Client --- metrics = { "cpu": 0, "ram": 0, "disk": 0, "cpu_model": "Remote Linux CPU", "cpu_cores": 2, "cpu_threads": 4, "ram_used_gb": 0.0, "ram_total_gb": 8.0, "ram_slots": [], "disk_used_gb": 0.0, "disk_total_gb": 80.0, "disk_part_number": "N/A", "disk_serial_number": "N/A", "service_tag": "N/A", "traffic_in": "0 B/s", "traffic_out": "0 B/s", "processes": [] } st_cmd = "(cat /sys/class/dmi/id/product_serial || cat /sys/devices/virtual/dmi/id/product_serial" if self.password: st_cmd += f" || echo '{self.password}' | sudo -S dmidecode -s system-serial-number" st_cmd += ") 2>/dev/null" # Eksekusi script gabungan di target Linux demi efisiensi tinggi compound_cmd = ( "echo '===CPU_MODEL==='; cat /proc/cpuinfo | grep -m 1 'model name' | cut -d: -f2-;" "echo '===CORES==='; grep -c '^processor' /proc/cpuinfo; grep -m 1 'cpu cores' /proc/cpuinfo | cut -d: -f2-;" "echo '===DISK==='; df -P / | tail -n 1;" "echo '===MEM==='; cat /proc/meminfo | grep -E 'MemTotal|MemAvailable|MemFree|Buffers|Cached';" "echo '===PROC==='; ps -eo pid,comm,%cpu,%mem --sort=-%cpu | head -n 7;" "echo '===TRAFFIC==='; cat /proc/stat | grep '^cpu '; cat /proc/net/dev;" f"echo '===SERVICE_TAG==='; {st_cmd};" "echo '===END==='" ) raw_out, _ = self.run_ssh_command(compound_cmd) if not raw_out: # Fallback jika SSH gagal/terputus self.is_connected = False return metrics self.is_connected = True try: # Parsing output compound secara terstruktur sections = {} current_section = None current_lines = [] for line in raw_out.split("\n"): line = line.strip() if line.startswith("===") and line.endswith("==="): if current_section: sections[current_section] = current_lines current_section = line.replace("===", "") current_lines = [] else: if line: current_lines.append(line) if current_section: sections[current_section] = current_lines # 1. CPU Model if "CPU_MODEL" in sections and sections["CPU_MODEL"]: metrics["cpu_model"] = sections["CPU_MODEL"][0].strip() # 2. Cores & Threads if "CORES" in sections and len(sections["CORES"]) >= 1: metrics["cpu_threads"] = int(sections["CORES"][0]) metrics["cpu_cores"] = int(sections["CORES"][1]) if len(sections["CORES"]) >= 2 else metrics["cpu_threads"] # 2.5. Service Tag if "SERVICE_TAG" in sections and sections["SERVICE_TAG"]: metrics["service_tag"] = sections["SERVICE_TAG"][0].strip() # 3. Disk Capacity if "DISK" in sections and sections["DISK"]: disk_line = sections["DISK"][0] parts = disk_line.split() if len(parts) >= 5: total_kb = float(parts[1]) used_kb = float(parts[2]) metrics["disk_total_gb"] = round(total_kb / (1024*1024), 1) metrics["disk_used_gb"] = round(used_kb / (1024*1024), 1) metrics["disk"] = int((used_kb / total_kb) * 100) # 4. Memory details (MemTotal & MemAvailable / MemFree) if "MEM" in sections and sections["MEM"]: mem_map = {} for l in sections["MEM"]: parts = l.split(":") if len(parts) >= 2: name = parts[0].strip() val = float(re.sub(r'[^0-9]', '', parts[1])) mem_map[name] = val total_kb = mem_map.get("MemTotal", 4194304.0) # Gunakan MemAvailable jika ada, jika tidak estimasikan dari MemFree + Cached available_kb = mem_map.get("MemAvailable", mem_map.get("MemFree", 0.0) + mem_map.get("Buffers", 0.0) + mem_map.get("Cached", 0.0)) used_kb = total_kb - available_kb metrics["ram_total_gb"] = round(total_kb / (1024*1024), 1) metrics["ram_used_gb"] = round(used_kb / (1024*1024), 1) metrics["ram"] = int((used_kb / total_kb) * 100) # 7. Membaca spesifikasi RAM asli menggunakan dmidecode via sudo (jika diizinkan) ram_slots = [] if self.password: try: cmd_dmidecode = f"echo '{self.password}' | sudo -S dmidecode -t memory 2>/dev/null" dmi_out, _ = self.run_ssh_command(cmd_dmidecode) if dmi_out and "Memory Device" in dmi_out: devices_raw = dmi_out.split("Memory Device") slot_index = 1 for dev_block in devices_raw[1:]: lines = dev_block.split("\n") dev_info = {} for line in lines: if ":" in line: k, v = line.split(":", 1) dev_info[k.strip()] = v.strip() size = dev_info.get("Size", "") if size and size != "No Module Installed" and "Size" not in size: locator = dev_info.get("Locator", f"Slot {slot_index}") m_type = dev_info.get("Type", "DDR4") speed = dev_info.get("Speed", "") if speed and speed != "Unknown": detail = f"{size} {m_type} @ {speed}" else: detail = f"{size} {m_type}" pn = dev_info.get("Part Number", "Not Specified") sn = dev_info.get("Serial Number", "Not Specified") ram_slots.append({ "name": f"Slot {slot_index} ({locator})", "detail": detail, "part_number": pn.strip() if pn else "Not Specified", "serial_number": sn.strip() if sn else "Not Specified" }) slot_index += 1 except Exception as e: print(f"Error parsing dmidecode memory: {e}") if not ram_slots: # RAM Slots fallback logic berdasarkan ukuran RAM r_tot = metrics["ram_total_gb"] if r_tot <= 4.2: ram_slots = [ {"name": "Slot 1", "detail": "4 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"} ] elif r_tot <= 8.2: ram_slots = [ {"name": "Slot 1", "detail": "8 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"} ] elif r_tot <= 16.2: ram_slots = [ {"name": "Slot 1", "detail": "8 GB DDR4 @ 2666MHz", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "8 GB DDR4 @ 2666MHz", "part_number": "N/A", "serial_number": "N/A"} ] else: ram_slots = [ {"name": "Slot 1", "detail": "16 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 2", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"}, {"name": "Slot 3", "detail": "16 GB DDR4", "part_number": "N/A", "serial_number": "N/A"}, {"name": "Slot 4", "detail": "Kosong", "part_number": "Kosong", "serial_number": "Kosong"} ] metrics["ram_slots"] = ram_slots # 8. Membaca model & serial harddisk fisik menggunakan lsblk try: disk_out, _ = self.run_ssh_command("lsblk -d -o MODEL,SERIAL") if disk_out: lines = [l.strip() for l in disk_out.split("\n") if l.strip()] if len(lines) >= 2: for line in lines[1:]: if "MODEL" in line: continue words = line.split() if len(words) >= 2: metrics["disk_serial_number"] = words[-1].strip() metrics["disk_part_number"] = " ".join(words[:-1]).strip() break elif len(words) == 1: metrics["disk_part_number"] = words[0].strip() break except Exception as e: print(f"Error parsing lsblk disk details: {e}") # 5. Top Processes if "PROC" in sections and sections["PROC"]: for line in sections["PROC"]: parts = line.split(None, 3) if len(parts) >= 4: # Abaikan baris header ps if not parts[0].strip().isdigit(): continue try: metrics["processes"].append({ "pid": parts[0], "name": parts[1], "cpu": float(parts[2]), "mem": parts[3] + " %" }) except ValueError: continue # 6. CPU & Network Traffic (Dihitung dari selisih poll-to-poll) if "TRAFFIC" in sections: try: cpu_line = [l for l in sections["TRAFFIC"] if l.startswith("cpu ")][0] c_parts = [float(x) for x in cpu_line.split()[1:]] total = sum(c_parts[:7]) idle = c_parts[3] + c_parts[4] # idle + iowait cur_time = time.time() prev_time = self.prev_traffic["time"] # Kalkulasi beban CPU if hasattr(self, "prev_cpu") and self.prev_cpu["time"] > 0.0: diff_total = total - self.prev_cpu["total"] diff_idle = idle - self.prev_cpu["idle"] if diff_total > 0: metrics["cpu"] = int(100 * (1.0 - (diff_idle / diff_total))) self.prev_cpu = {"time": cur_time, "total": total, "idle": idle} except Exception as e: print(f"Error parsing CPU Linux: {e}") try: # Kalkulasi kecepatan trafik dev_lines = [l for l in sections["TRAFFIC"] if not l.startswith("cpu ") and ":" in l] def sum_net_bytes(lines): sum_rx, sum_tx = 0, 0 for l in lines: parts = l.split(":", 1) if len(parts) >= 2: ifname = parts[0].strip() if ifname != "lo" and not re.search(r'docker|veth|br-|virbr|vmnet|vboxnet|vfb|tunnel', ifname): fields = parts[1].split() if len(fields) >= 9: sum_rx += int(fields[0]) sum_tx += int(fields[8]) return sum_rx, sum_tx rx, tx = sum_net_bytes(dev_lines) cur_time = time.time() prev_time = self.prev_traffic["time"] if prev_time > 0.0: interval = cur_time - prev_time if interval > 0: rx_speed = (rx - self.prev_traffic["rx"]) / interval tx_speed = (tx - self.prev_traffic["tx"]) / interval if rx_speed >= 0 and tx_speed >= 0: metrics["traffic_in"] = self.format_speed_bytes(rx_speed) metrics["traffic_out"] = self.format_speed_bytes(tx_speed) self.prev_traffic["time"] = cur_time self.prev_traffic["rx"] = rx self.prev_traffic["tx"] = tx except Exception as e: print(f"Error parsing network traffic: {e}") except Exception as e: print(f"Error parsing SSH metrics core: {e}") return metrics def execute_command(self, cmd_str): """Menjalankan perintah terminal kustom secara aman pada server.""" allowed_keywords = ["ping", "df", "free", "uptime", "uname", "systeminfo", "dir", "ls", "netstat", "ipconfig", "ifconfig"] clean_cmd = cmd_str.strip().lower() base_cmd = clean_cmd.split()[0] if clean_cmd else "" if base_cmd not in allowed_keywords: return f"Error: Perintah '{base_cmd}' diblokir demi keamanan. Anda hanya diizinkan menggunakan: {', '.join(allowed_keywords)}." # Pastikan perintah ping memiliki batas (count) agar tidak menggantung selamanya if base_cmd == "ping": is_linux = (self.server_type == "simulated") or (self.server_type == "ssh" and self.os_type == "linux") if is_linux: if "-c" not in clean_cmd: cmd_str += " -c 4" else: if "-n" not in clean_cmd and "-t" not in clean_cmd: cmd_str += " -n 4" if self.server_type == "simulated": if "ping" in clean_cmd: return f"Pinging {cmd_str.split()[-1]} dengan 32 byte data:\nBalasan dari {cmd_str.split()[-1]}: bytes=32 waktu=5ms TTL=64" elif "df" in clean_cmd or "dir" in clean_cmd or "ls" in clean_cmd: return "Filesystem Size Used Avail Use% Mounted on\n/dev/sda1 960G 748G 212G 78% /\ntmpfs 16G 0 16G 0% /dev/shm" elif "uptime" in clean_cmd: return " 08:29:12 up 14 days, 2:04, 2 users, load average: 0.15, 0.12, 0.08" else: return f"Simulasi Terminal: Perintah '{cmd_str}' berhasil dieksekusi dengan kode keluar 0." if self.server_type == "local": try: startupinfo = None if os.name == 'nt': startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW res = subprocess.run( cmd_str, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, startupinfo=startupinfo, timeout=4.0 ) return res.stdout or res.stderr or "Perintah selesai dijalankan tanpa keluaran." except Exception as e: return f"Error: {e}" # SSH Host stdout, stderr = self.run_ssh_command(cmd_str) return stdout or stderr or "Gagal mengeksekusi perintah remote."