refactor(report): Cleanup reports code

Author: gfieni

src/powerapi/report/control_report.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021, INRIA
+# Copyright (c) 2021, Inria
 # Copyright (c) 2021, University of Lille
 # All rights reserved.
 #
@@ -41,21 +41,21 @@ class ControlReport(Report):
     This is useful to control external tools via a producer/consumer job queue.
     """
 
-    def __init__(self, timestamp: datetime, sensor: str, target: str, action: str, parameters: list, metadata: dict[str, Any] = {}):
+    def __init__(self, timestamp: datetime, sensor: str, target: str, action: str, parameters: list, metadata: dict[str, Any] | None = None):
         """
-        Initialize a Control Event report using the given parameters.
         :param timestamp: Report timestamp
         :param sensor: Sensor name
         :param target: Target name
         :param action: Action name
         :param parameters: Parameter values
         """
-        Report.__init__(self, timestamp, sensor, target, metadata)
+        super().__init__(timestamp, sensor, target, metadata)
+
         self.action = action
         self.parameters = parameters
 
     def __repr__(self) -> str:
-        return f'ControlReport({self.timestamp}, {self.sensor}, {self.target}, {self.action}, {self.parameters}, {self.metadata})'
+        return f'ControlReport({self.timestamp}, {self.sensor}, {self.target}, {self.action}, {self.parameters})'
 
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, ControlReport):
@@ -83,4 +83,4 @@ def to_mongodb(report: ControlReport) -> dict:
         """
         :return: a dictionary, that can be stored into a mongodb, from a given ControlReport
         """
-        return report.__dict__
+        return vars(report)

src/powerapi/report/formula_report.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2022, INRIA
+# Copyright (c) 2022, Inria
 # Copyright (c) 2022, University of Lille
 # All rights reserved.
 #
@@ -44,18 +44,14 @@ class FormulaReport(Report):
     This is useful to gather information about a running formula in order to debug or compute statistics.
     """
 
-    def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: dict[str, Any]):
+    def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: dict[str, Any] | None = None):
         """
-        Initialize a Formula report using the given parameters.
         :param timestamp: Report timestamp
         :param sensor: Sensor name
         :param target: Target name
         :param metadata: Metadata values, can be anything but should be json serializable
         """
-        Report.__init__(self, timestamp, sensor, target, metadata)
-
-    def __repr__(self) -> str:
-        return f'FormulaReport({self.timestamp}, {self.sensor}, {self.target}, {self.metadata})'
+        super().__init__(timestamp, sensor, target, metadata)
 
     @staticmethod
     def to_csv_lines(report: FormulaReport, **_) -> (list[str], dict[str, Any]):
@@ -100,3 +96,10 @@ def to_influxdb(report: FormulaReport, **_) -> dict[str, Any]:
         }
 
         return document
+
+    @staticmethod
+    def to_json(report: FormulaReport) -> dict[str, Any]:
+        """
+        Convert a Formula report into a json document that can be stored into json.
+        """
+        return vars(report)

src/powerapi/report/hwpc_report.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021, INRIA
+# Copyright (c) 2021, Inria
 # Copyright (c) 2021, University of Lille
 # All rights reserved.
 #
@@ -69,22 +69,19 @@ class HWPCReport(Report):
         }
     """
 
-    def __init__(self, timestamp: datetime, sensor: str, target: str, groups: dict[str, dict],
-                 metadata: dict[str, Any] = {}):
+    def __init__(self, timestamp: datetime, sensor: str, target: str, groups: dict[str, dict], metadata: dict[str, Any] | None = None):
         """
-        Initialize an HWPC report using the given parameters.
-        :param datetime timestamp: Timestamp of the report
-        :param str sensor: Sensor name
-        :param str target: Target name
-        :param Dict groups: Events groups
+        :param timestamp: Timestamp of the report
+        :param sensor: Sensor name
+        :param target: Target name
+        :param groups: Events groups
         """
-        Report.__init__(self, timestamp, sensor, target, metadata)
+        super().__init__(timestamp, sensor, target, metadata)
 
-        #: (dict): Events groups
         self.groups = groups
 
     def __repr__(self) -> str:
-        return f'HWPCReport({self.timestamp}, {self.sensor}, {self.target}, {sorted(self.groups.keys())}, {self.metadata})'
+        return f'HWPCReport({self.timestamp}, {self.sensor}, {self.target}, {sorted(self.groups.keys())})'
 
     def __eq__(self, other) -> bool:
         if not isinstance(other, HWPCReport):
@@ -112,7 +109,7 @@ def from_json(data: dict) -> HWPCReport:
 
     @staticmethod
     def to_json(report: HWPCReport) -> dict:
-        return report.__dict__
+        return vars(report)
 
     @staticmethod
     def from_mongodb(data: dict) -> HWPCReport:

src/powerapi/report/power_report.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021, INRIA
+# Copyright (c) 2021, Inria
 # Copyright (c) 2021, University of Lille
 # All rights reserved.
 #
@@ -42,21 +42,20 @@ class PowerReport(Report):
     PowerReport stores the power estimation information.
     """
 
-    def __init__(self, timestamp: datetime, sensor: str, target: str, power: float, metadata: dict[str, Any] = {}):
+    def __init__(self, timestamp: datetime, sensor: str, target: str, power: float, metadata: dict[str, Any] | None = None):
         """
-        Initialize a Power report using the given parameters.
         :param datetime timestamp: Report timestamp
         :param str sensor: Sensor name
         :param str target: Target name
         :param float power: Power value
         :param dict metadata: Metadata values, can be anything that add useful information
         """
-        Report.__init__(self, timestamp, sensor, target, metadata)
+        super().__init__(timestamp, sensor, target, metadata)
 
         self.power = power
 
     def __repr__(self) -> str:
-        return f'PowerReport({self.timestamp}, {self.sensor}, {self.target}, {self.power}, {self.metadata})'
+        return f'PowerReport({self.timestamp}, {self.sensor}, {self.target}, {self.power})'
 
     def __eq__(self, other) -> bool:
         if not isinstance(other, PowerReport):

src/powerapi/report/report.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021, INRIA
+# Copyright (c) 2021, Inria
 # Copyright (c) 2021, University of Lille
 # All rights reserved.
 #
@@ -35,7 +35,7 @@
 from typing import NewType, Any
 from zlib import crc32
 
-from powerapi.exception import PowerAPIExceptionWithMessage, PowerAPIException
+from powerapi.exception import PowerAPIExceptionWithMessage
 from powerapi.message import Message
 
 TIMESTAMP_KEY = 'timestamp'
@@ -60,59 +60,32 @@ def __init__(self, msg, input_data):
         self.input_data = input_data
 
 
-class DeserializationFail(PowerAPIException):
-    """
-    Exception raised when the
-    in the good format
-    """
-
-
 class Report(Message):
     """
-    Report abtract class.
+    Report abstract class.
     """
 
-    def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: dict[str, Any] = {}):
+    def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: dict[str, Any] | None = None):
         """
-        Initialize a report using the given parameters.
-        :param datetime timestamp: Timestamp
-        :param str sensor: Sensor name.
-        :param str target: Target name.
+        :param timestamp: Timestamp of the measurements.
+        :param sensor: Name of the sensor from which the measurements were taken.
+        :param target: Name of the target that the measurements refer to.
         """
-        Message.__init__(self, None)
         self.timestamp = timestamp
         self.sensor = sensor
         self.target = target
-        self.metadata = dict(metadata)
-
-        #: id given by the dispatcher actor in order manage report order
-        self.dispatcher_report_id = None
-
-    def __str__(self):
-        return f'{self.__class__.__name__}({self.timestamp}, {self.sensor}, {self.target}, {self.metadata})'
+        self.metadata = metadata if metadata is not None else {}
 
-    def __repr__(self):
-        return f'{self.__class__.__name__}({self.timestamp}, {self.sensor}, {self.target}, {self.metadata})'
+    def __repr__(self) -> str:
+        return f'{self.__class__.__name__}({self.timestamp}, {self.sensor}, {self.target})'
 
-    def __eq__(self, other):
+    def __eq__(self, other) -> bool:
         return (isinstance(other, type(self)) and
                 self.timestamp == other.timestamp and
                 self.sensor == other.sensor and
                 self.target == other.target and
                 self.metadata == other.metadata)
 
-    @staticmethod
-    def to_json(report: Report) -> dict:
-        """
-        :return: a json dictionary, that can be converted into json format, from a given Report
-        """
-        json = report.__dict__
-        # sender_name and dispatcher_report_id are not used
-        json.pop('sender_name')
-        json.pop('dispatcher_report_id')
-
-        return json
-
     @staticmethod
     def _extract_timestamp(ts):
         # Unix timestamp format (in milliseconds)