Add statistics file to telemetry with usage count.

src/backend/backend.py

@@ -100,3 +100,9 @@ def cid_file_initialized(self):
 
         :return: True if client ID file is initialized, otherwise False.
         """
+
+    @abc.abstractmethod
+    def set_stats(self, data: dict):
+        """
+        Pass additional statistic, which will be added to telemetry messages
+        """

src/backend/backend_ga.py

@@ -115,6 +115,9 @@ def generate_new_cid_file(self):
     def cid_file_initialized(self):
         return self.cid is not None
 
+    def set_stats(self, data: dict):
+        pass
+
 
 def is_valid_cid(cid: str):
     try:

src/backend/backend_ga4.py

@@ -40,6 +40,7 @@ def __init__(self, tid: str = None, app_name: str = None, app_version: str = Non
             'app_name': self.app_name,
             'app_version': self.app_version,
         }
+        self.stats = {}
 
     def send(self, message: dict):
         if message is None:
@@ -90,6 +91,7 @@ def build_event_message(self, event_category: str, event_action: str, event_labe
                         "event_count": event_value,
                         "session_id": self.session_id,
                         **default_args,
+                        **self.stats
                     }
                 }
             ]
@@ -124,6 +126,9 @@ def remove_cid_file(self):
         remove_cid_file(self.cid_filename)
         remove_cid_file(self.old_cid_filename)
 
+    def set_stats(self, data: dict):
+        self.stats = data
+
 
 def is_valid_cid(cid: str):
     try:

src/main.py

@@ -7,8 +7,9 @@
 from enum import Enum
 
 from .backend.backend import BackendRegistry
-from .utils.opt_in_checker import OptInChecker, ConsentCheckResult, DialogResult
 from .utils.sender import TelemetrySender
+from .utils.opt_in_checker import OptInChecker, ConsentCheckResult, DialogResult
+from .utils.stats_processor import StatsProcessor
 
 
 class OptInStatus(Enum):
@@ -76,6 +77,9 @@ def init(self, app_name: str = None, app_version: str = None, tid: str = None,
         if self.consent and not self.backend.cid_file_initialized():
             self.backend.generate_new_cid_file()
 
+        if self.consent:
+            self.backend.set_stats(self.get_stats())
+
         if not enable_opt_in_dialog and self.consent:
             # Try to create directory for client ID if it does not exist
             if not opt_in_checker.create_or_check_consent_dir():
@@ -264,6 +268,8 @@ def _update_opt_in_status(tid: str, new_opt_in_status: bool):
             if prev_status != OptInStatus.DECLINED:
                 telemetry.send_opt_in_event(OptInStatus.DECLINED, prev_status, force_send=True)
             telemetry.backend.remove_cid_file()
+            from .utils.stats_processor import StatsProcessor
+            StatsProcessor().remove_stats_file()
             print("You have successfully opted out to send the telemetry data.")
 
     def send_opt_in_event(self, new_state: OptInStatus, prev_state: OptInStatus = OptInStatus.UNDEFINED,
@@ -283,6 +289,24 @@ def send_opt_in_event(self, new_state: OptInStatus, prev_state: OptInStatus = Op
             label = "{{prev_state:{}, new_state: {}}}".format(prev_state.value, new_state.value)
             self.send_event("opt_in", new_state.value, label, force_send=force_send)
 
+    def get_stats(self):
+        stats = StatsProcessor()
+        file_exists, data = stats.get_stats()
+        if not file_exists:
+            created = stats.create_new_stats_file()
+            data = {}
+            if not created:
+                return None
+        if "usage_count" in data:
+            usage_count = data["usage_count"]
+            if usage_count < sys.maxsize:
+                usage_count += 1
+        else:
+            usage_count = 1
+        data["usage_count"] = usage_count
+        stats.update_stats(data)
+        return data
+
     @staticmethod
     def opt_in(tid: str):
         """

src/main_test.py

@@ -72,7 +72,8 @@ def make_message(self, client_id, app_name, app_version, category, action, label
                                          'event_count': value,
                                          'session_id': session_id,
                                          'app_name': app_name,
-                                         'app_version': app_version}
+                                         'app_version': app_version,
+                                         'usage_count': 1}
                                     }
                                ]}
 

src/utils/stats_processor.py

@@ -0,0 +1,81 @@
+# Copyright (C) 2018-2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from .opt_in_checker import OptInChecker
+import logging as log
+import os
+import json
+
+
+class StatsProcessor:
+    def __init__(self):
+        self.opt_in_checker = OptInChecker()
+
+    def stats_file(self):
+        """
+        Returns the statistics file path.
+        """
+        return os.path.join(self.opt_in_checker.consent_file_base_dir(), self.opt_in_checker.consent_file_subdirectory(), "stats")
+
+    def create_new_stats_file(self):
+        """
+        Creates a new statistics file.
+        :return: True if the file is created successfully, otherwise False
+        """
+        if not self.opt_in_checker.create_or_check_consent_dir():
+            return False
+        try:
+            open(self.stats_file(), 'w').close()
+        except Exception:
+            return False
+        return True
+
+    def update_stats(self, stats: dict):
+        """
+        Updates the statistics in the statistics file.
+        :param stats: the dictionary with statistics.
+        :return: False if the statistics file is not writable, otherwise True
+        """
+        if self.opt_in_checker.consent_file_base_dir() is None or self.opt_in_checker.consent_file_subdirectory() is None:
+            return False
+        if not os.path.exists(self.stats_file()):
+            if not self.create_new_stats_file():
+                return False
+        if not os.access(self.stats_file(), os.W_OK):
+            log.warning("Failed to usage statistics. "
+                        "Please allow write access to the following file: {}".format(self.stats_file()))
+            return False
+        try:
+            str_data = json.dumps(stats, indent=4)
+            with open(self.stats_file(), 'w') as file:
+                file.write(str_data)
+        except Exception:
+            return False
+        return True
+
+    def get_stats(self):
+        """
+        Gets information from statistics file.
+        :return: the tuple, where the first element is True if the file is read successfully, otherwise False
+        and the second element is the content of the statistics file.
+        """
+        if not os.access(self.stats_file(), os.R_OK):
+            return False, {}
+        try:
+            with open(self.stats_file(), 'r') as file:
+                data = json.load(file)
+        except Exception:
+            return False, {}
+        return True, data
+
+    def remove_stats_file(self):
+        """
+        Removes statistics file.
+        :return: None
+        """
+        stats_file = self.stats_file()
+        if os.path.exists(stats_file):
+            if not os.access(stats_file, os.W_OK):
+                log.warning("Failed to remove statistics file {}.".format(stats_file))
+                return
+            os.remove(stats_file)

src/utils/stats_processor_test.py

@@ -0,0 +1,65 @@
+# Copyright (C) 2018-2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+import unittest
+from tempfile import TemporaryDirectory
+from unittest.mock import MagicMock, patch
+
+from .stats_processor import StatsProcessor
+from .opt_in_checker import OptInChecker
+
+
+class StatsProcessorTest(unittest.TestCase):
+    test_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'test_stats')
+    test_subdir = 'test_stats_subdir'
+    stats_processor = StatsProcessor()
+
+    def init_stats_processor(self, test_directory):
+        self.stats_processor.consent_file_base_dir = MagicMock(return_value=test_directory)
+        self.stats_processor.consent_file_subdirectory = MagicMock(return_value=self.test_subdir)
+
+    def test_stats_usage(self):
+        with TemporaryDirectory(prefix=self.test_directory) as test_dir:
+            with patch.object(OptInChecker, 'consent_file_base_dir', return_value=test_dir):
+                with patch.object(OptInChecker, 'consent_file_subdirectory', return_value=self.test_subdir):
+                    if not os.path.exists(test_dir):
+                        os.mkdir(test_dir)
+                    stats_filename = os.path.join(test_dir, self.test_subdir, 'stats')
+                    test_data1 = {"value1": 12, "value2": 7, "value3": 8}
+
+                    # Test first creation of statistics file
+                    self.stats_processor.update_stats(test_data1)
+                    self.assertTrue(os.path.exists(stats_filename))
+                    with open(stats_filename, 'r') as file:
+                        self.assertTrue(file.readlines() == ['{\n', '    "value1": 12,\n', '    "value2": 7,\n', '    "value3": 8\n', '}'])
+
+                    status, res = self.stats_processor.get_stats()
+                    self.assertTrue(status)
+                    self.assertTrue(res == test_data1)
+
+                    # Test updating of statistics file
+                    test_data2 = {"value1": 15, "a": "abs"}
+                    self.stats_processor.update_stats(test_data2)
+                    self.assertTrue(os.path.exists(stats_filename))
+                    with open(stats_filename, 'r') as file:
+                        self.assertTrue(file.readlines() == ['{\n', '    "value1": 15,\n', '    "a": "abs"\n', '}'])
+
+                    status, res = self.stats_processor.get_stats()
+                    self.assertTrue(status)
+                    self.assertTrue(res == test_data2)
+
+                    # Test removing of statistics file
+                    self.stats_processor.remove_stats_file()
+                    self.assertFalse(os.path.exists(stats_filename))
+
+                    status, res = self.stats_processor.get_stats()
+                    self.assertFalse(status)
+                    self.assertTrue(res == {})
+
+                    # Test attempt to read incorrect statistics file
+                    with open(stats_filename, 'w') as file:
+                        file.write("{ abc")
+                    status, res = self.stats_processor.get_stats()
+                    self.assertFalse(status)
+                    self.assertTrue(res == {})