Add support for liquid cooling of mellnox api

Signed-off-by: Yuanzhe, Liu <[email protected]>

Author: yuazhe

dockers/docker-platform-monitor/docker-pmon.supervisord.conf.j2

@@ -218,6 +218,14 @@ dependent_startup_wait_for=rsyslogd:running {% if delay_non_critical_daemon %}de
     {% if thermalctld.thermal_monitor_update_elapsed_threshold is defined and thermalctld.thermal_monitor_update_elapsed_threshold is not none %}
         {%- set options = options + " --thermal-monitor-update-elapsed-threshold " + thermalctld.thermal_monitor_update_elapsed_threshold|string %}
     {% endif -%}
+
+    {% if thermalctld.enable_liquid_cooling %}
+    {%- set options = options + " --enable_liquid_cooling" %}
+    {% endif -%}
+
+    {% if thermalctld.liquid_cooling_update_interval %}
+    {%- set options = options ~ " --liquid_cooling_update_interval " ~ thermalctld.liquid_cooling_update_interval %}
+    {% endif -%}
 {% endif -%}
 
 {%- set command = base_command ~ options %}

platform/mellanox/mlnx-platform-api/sonic_platform/chassis.py

@@ -121,10 +121,13 @@ def __init__(self):
         self._RJ45_port_inited = False
         self._RJ45_port_list = None
 
+
         # Build the CPO port list from platform.json and hwsku.json
         self._cpo_port_inited = False
         self._cpo_port_list = None
 
+        self.liquid_cooling = None
+
         Chassis.chassis_instance = self
 
         self.module_host_mgmt_initializer = module_host_mgmt_initializer.ModuleHostMgmtInitializer()
@@ -1118,6 +1121,20 @@ def is_replaceable(self):
         """
         return False
 
+    
+    ##############################################
+    # LiquidCooling methods
+    ##############################################
+
+    def initialize_liquid_cooling(self):
+        if not self.liquid_cooling:
+            from .liquid_cooling import LiquidCooling
+            self.liquid_cooling = LiquidCooling()
+
+    def get_liquid_cooling(self):
+        self.initialize_liquid_cooling()
+        return self.liquid_cooling
+
 
 class ModularChassis(Chassis):
     def __init__(self):

platform/mellanox/mlnx-platform-api/sonic_platform/liquid_cooling.py

@@ -0,0 +1,75 @@
+#
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2019-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#############################################################################
+# Mellanox
+#
+# Module contains an implementation of SONiC Platform Base API and
+# provides the liquid cooling status which are available in the platform
+#
+#############################################################################
+
+try:
+    from sonic_platform_base.liquid_cooling_base import LeakageSensorBase,LiquidCoolingBase
+    from sonic_py_common.logger import Logger
+    from . import utils
+    import os
+    import glob
+    import logging
+except ImportError as e:
+    raise ImportError(str(e) + "- required module not found")
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+LIQUID_COOLING_SENSOR_PATH = "/var/run/hw-management/system/"
+
+class LeakageSensor(LeakageSensorBase):
+    def __init__(self, name,path):
+        super(LeakageSensor, self).__init__(name)
+        self.path = path
+
+    def is_leak(self):
+        content = utils.read_int_from_file(self.path, 'N/A')
+
+        if content == 1:
+            self.leaking = False
+            return False
+        elif content == 0:
+            self.leaking = True
+            return True
+        else:
+            logger.error(f"Failed to read leakage sensor {self.name} value: {content}")
+            return 'N/A'
+
+class LiquidCooling(LiquidCoolingBase):
+    """Platform-specific Liquid Cooling class"""
+
+    def __init__(self):
+        
+        self.leakage_sensors_num = utils.read_int_from_file("/var/run/hw-management/config/leakage_counter")
+        self.leakage_sensors = []
+
+        sensor_files = glob.glob(os.path.join(LIQUID_COOLING_SENSOR_PATH, "leakage*"))
+        sensor_files.sort(key=lambda x: int(x.split("leakage")[-1]))
+
+        for sensor_path in sensor_files[:self.leakage_sensors_num]:
+            sensor_name = os.path.basename(sensor_path)
+            self.leakage_sensors.append(LeakageSensor(sensor_name, sensor_path))
+
+        super(LiquidCooling, self).__init__(self.leakage_sensors_num, self.leakage_sensors)
+

platform/mellanox/mlnx-platform-api/tests/test_liquid_cooling.py

@@ -0,0 +1,55 @@
+import os
+import sys
+from unittest import mock
+from sonic_platform.liquid_cooling import LiquidCooling, LeakageSensor
+from sonic_platform.utils import read_int_from_file
+
+def test_leakage_sensor_init():
+    sensor = LeakageSensor("leakage1", "/test/path")
+    assert sensor.name == "leakage1"
+    assert sensor.path == "/test/path"
+
+def test_leakage_sensor_is_leak():
+    sensor = LeakageSensor("leakage1", "/test/path")
+    
+    # Test when file exists and content is "1" (no leak)
+    with mock.patch('os.path.exists') as mock_exists:
+        with mock.patch('builtins.open', mock.mock_open(read_data="1")):
+            mock_exists.return_value = True
+            assert sensor.is_leak() is False
+
+    # Test when file exists and content is "0" (leak detected)
+    with mock.patch('os.path.exists') as mock_exists:
+        with mock.patch('builtins.open', mock.mock_open(read_data="0")):
+            mock_exists.return_value = True
+            assert sensor.is_leak() is True
+
+    # Test when file does not exist
+    with mock.patch('os.path.exists') as mock_exists:
+        mock_exists.return_value = False
+        assert sensor.is_leak() == 'N/A'
+
+def test_liquid_cooling_init():
+    with mock.patch('os.path.exists') as mock_exists, \
+         mock.patch('os.path.join', side_effect=lambda *args: "/".join(args)) as mock_join, \
+         mock.patch('glob.glob') as mock_glob, \
+         mock.patch('sonic_platform.utils.read_int_from_file', return_value=4) as mock_read_int:
+        
+        # Setup mock to simulate 4 leakage sensors
+        mock_exists.side_effect = [True, True, True, True]
+        mock_glob.return_value = [
+            "/var/run/hw-management/system/leakage1",
+            "/var/run/hw-management/system/leakage2", 
+            "/var/run/hw-management/system/leakage3",
+            "/var/run/hw-management/system/leakage4"
+        ]
+
+        liquid_cooling = LiquidCooling()
+        
+        # Verify the number of sensors initialized
+        assert liquid_cooling.get_num_leak_sensors() == 4
+        
+
+        with mock.patch.object(liquid_cooling.leakage_sensors[3], 'is_leak', return_value=False) as mock_sensor1:
+            sensors = liquid_cooling.get_leak_sensor_status()
+            assert len(sensors) == 3