Index: dialin/dg/SampleWater.py
===================================================================
diff -u
--- dialin/dg/SampleWater.py	(revision 0)
+++ dialin/dg/SampleWater.py	(revision a669aec0522f546a2e40b81d9b55c8b87d37f7e9)
@@ -0,0 +1,80 @@
+###########################################################################
+#
+# Copyright (c) 2019-2021 Diality Inc. - All Rights Reserved.
+#
+# THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN
+# WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER.
+#
+# @file samplewater.py
+#
+# @author (last) Quang Nguyen
+# @date (last) 02-Mar-2021
+# @author (original) Quang Nguyen
+# @date (original) 02-Mar-2021
+#
+############################################################################
+import struct
+from ..protocols.CAN import (DenaliMessage,
+                             DenaliChannels)
+from ..utils.base import _AbstractSubSystem, _publish
+from ..common.msg_defs import MsgIds, MsgFieldPositions
+from logging import Logger
+
+class DGSampleWater(_AbstractSubSystem):
+    """
+
+    Dialysate Generator (DG) Dialin API sub-class for sample water related commands.
+
+    """
+
+    def __init__(self, can_interface, logger: Logger):
+        """
+        DGSampleWater constructor
+        """
+
+        super().__init__()
+        self.can_interface = can_interface
+        self.logger = logger
+
+        if self.can_interface is not None:
+            channel_id = DenaliChannels.hd_sync_broadcast_ch_id
+            msg_id = MsgIds.MSG_ID_DG_FILTER_FLUSH_PROGRESS.value
+            self.can_interface.register_receiving_publication_function(channel_id, msg_id,
+                                                                       self._handler_filter_flush_progress_sync)
+
+        self.filter_flush_timeout = 0
+        self.filter_flush_time_countdown = 0
+
+    def get_filter_flush_timeout(self):
+        """
+        Gets the filter flush timeout
+
+        @return: The filter flush timeout
+        """
+        return self.filter_flush_timeout
+
+    def get_filter_flush_time_countdown(self):
+        """
+        Gets the filter flush time countdown
+
+        @return: The filter flush time countdown
+        """
+        return self.filter_flush_time_countdown
+
+
+    @_publish([
+        "filter_flush_timeout",
+        "filter_flush_time_countdown"
+    ])
+    def _handler_filter_flush_progress_sync(self, message):
+        """
+        Handles published filter flush progress data messages. Filter flush progress data are captured for reference.
+
+        @param message: published filter flush progress data message
+        @return: None
+        """
+
+        self.filter_flush_timeout = struct.unpack('i', bytearray(
+            message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0]
+        self.filter_flush_time_countdown = struct.unpack('i', bytearray(
+            message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0]