convenience.py

import paramiko
import argparse
import sys
import time
import lib.ssh_util as ssh_util
import lib.bmc_boot_utils as bbu
from lib.init_logger import init_logger
import lib.bmc_util as bu
import random
import shutil
import os
from ac import ac

# for uart bic fw update path
loader_signed = "img/uart_bic_fw_update/loader_signed.bin"
mmc_recovery = "img/uart_bic_fw_update/mmc-recovery"

# fw map
fw_map = {
    "pcie": "/home/adam/meta/OpenBIC_project/meta_OpenBIC/pldm_fw_creator/switch/new.pldm",
    "bic": "/home/billy/Desktop/meta/sitv3/build/zephyr/SB_SI.pldm",
    "bic_bin": "/home/adam/meta/OpenBIC_project/meta_OpenBIC/build/zephyr/SB_SI_signed.bin",
    "vr_pex": "/home/adam/meta/OpenBIC_project/meta_OpenBIC/pldm_fw_creator/vr/pex.pldm",
    "vr_a1": "/home/adam/meta/OpenBIC_project/meta_OpenBIC/pldm_fw_creator/vr/a1.pldm",
}

CONSOLE_SERVICES = [
    "obmc-console@ttyUSBdevice1.service",
    "obmc-console@ttyUSBdevice15.service",
    "obmc-console@ttyUSBdevice29.service",
    "obmc-console@ttyUSBdevice43.service",
]

logger = init_logger("bmc_reboot.log", verbose=True)
logger.propagate = False

def dc_power_on(ssh):
    ssh.exec_command("i2cset -f -y 21 0x33 0x38 0x01")
    time.sleep(5)

def fw_update(ssh, ip, component="bic", remote_dir="/tmp"):
    target = {
        "ip": ip,
        "port": 22,
        "username": "root",
        "password": "0penBmc",
    }

    local_path = fw_map[component]
    file_name = os.path.basename(local_path)
    remote_path = f"{remote_dir}/{file_name}"

    # 1) push to /tmp (or remote_dir)
    if not ssh_util.push_file(target, local_path, remote_dir=remote_dir):
        logger.error(f"push file failed: {local_path} -> {remote_path}")
        return False

    # 2) verify file exists
    ssh_util.run_command(ssh, f"ls -l {remote_path}", logger)

    # 3) cd to remote_dir then run with relative filename (workaround)
    cmd = f"sh -lc 'cd {remote_dir} && pldm-update ag {file_name}'"
    ok = ssh_util.run_command(ssh, cmd, logger)
    if not ok:
        logger.error(f"fw update fail (cmd={cmd})")
        return False

    logger.info("fw update success")
    return True

def fw_update_uart(ssh, ip, port=3):
    target = {
        "ip": ip,
        "port": 22,
        "username": "root",
        "password": "0penBmc"
    }
    mmcs = ["tty_SITV_0", "tty_SITV_1", "tty_SITV_2", "tty_SITV_3"]
    mmc = bu.SerialSSHClient(ssh, mmcs[port], logger)
    mmc.send_command("erase")
    logger.info(f"mmc port {port} erased!!")
    mmc.close()

    import tarfile
    files = [fw_map["bic_bin"], loader_signed]
    output = "SB_SI_signed.tar.xz"
    with tarfile.open(output, "w:xz") as tar:
        for f in files:
            tar.add(f, arcname=f.split("/")[-1])
    logger.info(f"Created {output}")

    if not ssh_util.push_file(target, mmc_recovery, remote_dir="/tmp"):
        print("push 'mmc-recovery' file: Fail")
        return False
    bic_mmc_recovery = "/tmp/" + mmc_recovery.split('/')[-1]
    if not ssh_util.push_file(target, output, remote_dir="/tmp"):
        print("push '.tar.xz' file: Fail")
        return False
    os.remove(output)
    ssh.exec_command(f"chmod 777 {bic_mmc_recovery}")
    if  ssh_util.run_command(ssh, f"{bic_mmc_recovery} ag /tmp/{output} -n {port} -b sit", logger):
        ac(ssh, target)
        return True
    return False

def mctp_link(ssh, *args):
    command = f"""
    mctp link;
    mctp link set mctpi3c0 up;
    mctp addr;
    mctp addr add 8 dev mctpi3c0;
    """
    ssh.exec_command(command)
    time.sleep(0.1)
    for i, arg in enumerate(args):
        command = f"""
        busctl call au.com.codeconstruct.MCTP1 /au/com/codeconstruct/mctp1/interfaces/mctpi3c0 au.com.codeconstruct.MCTP.BusOwner1 \
            LearnEndpoint ay 6 0x06 0x32 0x12 0x34 0x55 0x{arg};
        """
        ssh_util.run_command(ssh, command, logger)
    ssh_util.run_command(ssh, "busctl tree au.com.codeconstruct.MCTP1", logger)

def mmc_log_record(ssh):
    mmcs = ["tty_SITV_0", "tty_SITV_1", "tty_SITV_2", "tty_SITV_3"]
    for i in range(len(mmcs)):
        mmcs[i] = bu.SerialSSHClient(ssh, mmcs[i], logger)
    while (input("enter 0 to exit : ") != "0"):
        for mmc in mmcs:
            logger.info(f"-------sitv {mmc.port[-1]}--------")
            # mmc.send_command("platform info")
            mmc.log_inf()
            mmc.clear_log()
    for mmc in mmcs:
        mmc.close()

def _systemctl_stop_services(ssh, logger, services):
    for svc in services:
        cmd = f"systemctl stop {svc}"
        logger.info(f"[BMC] {cmd}")
        stdin, stdout, stderr = ssh.exec_command(cmd)
        rc = stdout.channel.recv_exit_status()
        if rc != 0:
            err = (stderr.read() or b"").decode(errors="ignore").strip()
            logger.warning(f"stop failed: {svc} (rc={rc}) {err}")

def _systemctl_start_services(ssh, logger, services):
    for svc in services:
        cmd = f"systemctl start {svc}"
        logger.info(f"[BMC] {cmd}")
        stdin, stdout, stderr = ssh.exec_command(cmd)
        rc = stdout.channel.recv_exit_status()
        if rc != 0:
            err = (stderr.read() or b"").decode(errors="ignore").strip()
            logger.warning(f"start failed: {svc} (rc={rc}) {err}")

def mmc_log_record_start(
    ssh,
    logger,
    ports=(0, 1, 2, 3),
    console_services=CONSOLE_SERVICES,
    stop_console_before_start=True,
    clear_on_start=True,
):
    """
    Start MMC log recording session:
    - stop obmc-console services (optional)
    - open SerialSSHClient
    - optionally clear logs ONCE at start
    NOTE: This function will NOT print logs.
    Returns: mmcs (list of SerialSSHClient)
    """
    if stop_console_before_start and console_services:
        _systemctl_stop_services(ssh, logger, console_services)
        time.sleep(0.3)

    mmc_ports = ["tty_SITV_0", "tty_SITV_1", "tty_SITV_2", "tty_SITV_3"]

    mmcs = []
    for p in ports:
        if p < 0 or p >= len(mmc_ports):
            raise ValueError(f"Invalid MMC port: {p}")
        mmcs.append(bu.SerialSSHClient(ssh, mmc_ports[p], logger))

    logger.info(f"MMC log record START (no print): ports={ports}")

    if clear_on_start:
        for mmc in mmcs:
            try:
                mmc.clear_log()
            except Exception as e:
                logger.warning(f"clear_log failed on {getattr(mmc, 'port', 'unknown')}: {e}")

    return mmcs


def mmc_log_record_exit(
    mmcs,
    ssh,
    logger,
    console_services=CONSOLE_SERVICES,
    start_console_after_exit=True,
    dump_log_on_exit=True,
    clear_after_dump=False,
):
    """
    Exit MMC log recording session:
    - optionally dump logs ONCE at exit
    - close all SerialSSHClient
    - start obmc-console services back (optional)
    """
    if mmcs and dump_log_on_exit:
        logger.info("========== MMC LOG DUMP (ON EXIT) ==========")
        for mmc in mmcs:
            try:
                logger.info(f"-------sitv {mmc.port[-1]}--------")
                mmc.log_inf()
                if clear_after_dump:
                    mmc.clear_log()
            except Exception as e:
                logger.warning(f"log dump failed on {getattr(mmc, 'port', 'unknown')}: {e}")

    if mmcs:
        for mmc in mmcs:
            try:
                mmc.close()
            except Exception as e:
                logger.warning(f"close failed on {getattr(mmc, 'port', 'unknown')}: {e}")

    if start_console_after_exit and console_services:
        _systemctl_start_services(ssh, logger, console_services)

    logger.info("MMC log record EXIT: dumped-once + all ports closed")

def main(ip):
    ssh = ssh_util.connect_bmc(ip, logger)
    # mmc_log_record(ssh)
    # mctp_link(ssh, 40)
    mmcs = mmc_log_record_start(
        ssh, logger,
        ports=(0,1,2,3),
        clear_on_start=True
    )
    fw_update(ssh, ip, "bic")
    mmc_log_record_exit(mmcs, ssh, logger)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-ip", help="Target BMC IP address", default="10.10.12.37")  # Default IP set here
    args = parser.parse_args()
    main(args.ip)