Merge remote-tracking branch 'upstream/hotfixes' into release

Author: Javert899

pm4py/discovery.py

@@ -240,6 +240,10 @@ def discover_performance_dfg(
     :param timestamp_key: Attribute to be used for the timestamp (default: "time:timestamp").
     :param case_id_key: Attribute to be used as case identifier (default: "case:concept:name").
     :param perf_aggregation_key: Selector for the type of aggregation (all, mean, median, max, min, sum, stdev)
+    When business hours are enabled, the returned performance DFG remains
+    dictionary-compatible and also retains the configured schedule so that
+    visualizations can express its durations in working days.
+
     :return: A tuple of three dictionaries: (performance_dfg, start_activities, end_activities).
     :rtype: ``Tuple[dict, dict, dict]``
 
@@ -332,6 +336,13 @@ def discover_performance_dfg(
         end_activities = end_activities_module.get_end_activities(
             log, parameters=properties
         )
+    if business_hours:
+        from pm4py.objects.dfg.obj import PerformanceDFG
+
+        dfg = PerformanceDFG(
+            dfg, business_hour_slots=business_hour_slots
+        )
+
     return dfg, start_activities, end_activities
 
 

pm4py/objects/dfg/obj.py

@@ -23,6 +23,29 @@
 from typing import Tuple, Any, Counter as TCounter
 
 
+class PerformanceDFG(dict):
+    """Dictionary-compatible performance DFG with calculation metadata.
+
+    Performance values are intentionally kept as a regular mapping for
+    backwards compatibility.  The optional business-hour schedule lets
+    downstream consumers render those values using the same definition of a
+    working day that was used during discovery.
+    """
+
+    def __init__(self, *args, business_hour_slots=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.business_hour_slots = (
+            tuple(tuple(slot) for slot in business_hour_slots)
+            if business_hour_slots is not None
+            else None
+        )
+
+    def copy(self):
+        return type(self)(
+            self, business_hour_slots=self.business_hour_slots
+        )
+
+
 class DirectlyFollowsGraph:
 
     def __init__(self, graph=None, start_activities=None, end_activities=None):

pm4py/util/vis_utils.py

@@ -30,8 +30,12 @@
 MIN_EDGE_PENWIDTH_GRAPHVIZ = 1.0
 
 
-def get_business_hour_slots(parameters):
-    """Return the schedule to use for duration labels, if configured."""
+def get_business_hour_slots(parameters, performance_dfg=None):
+    """Return the schedule to use for duration labels, if configured.
+
+    Explicit visualization parameters take precedence over metadata retained
+    by a discovered performance DFG.
+    """
     from pm4py.util import constants, exec_utils
 
     missing = object()
@@ -43,8 +47,14 @@ def get_business_hour_slots(parameters):
     )
     if business_hours is not missing and not business_hours:
         return None
-    if business_hour_slots is not missing:
+    if business_hour_slots is not missing and business_hour_slots is not None:
         return business_hour_slots
+    if performance_dfg is not None:
+        discovered_slots = getattr(
+            performance_dfg, "business_hour_slots", None
+        )
+        if discovered_slots is not None:
+            return discovered_slots
     if business_hours is not missing and business_hours:
         return constants.DEFAULT_BUSINESS_HOUR_SLOTS
     return None

pm4py/visualization/dfg/variants/performance.py

@@ -125,7 +125,7 @@ def apply(
         parameters,
         "Performance Directly-Follows Graph",
     )
-    business_hour_slots = vis_utils.get_business_hour_slots(parameters)
+    business_hour_slots = vis_utils.get_business_hour_slots(parameters, dfg)
 
     # if all the aggregation measures are provided for a given key,
     # then pick one of the values for the representation

tests/business_hours_test.py

@@ -1,6 +1,7 @@
 import random
 import unittest
 from datetime import date, datetime, timedelta, timezone
+from unittest.mock import patch
 
 from pm4py.util.business_hours import (
     BusinessHours,
@@ -58,6 +59,49 @@ def test_performance_dfg_uses_business_days_in_labels(self):
 
         self.assertIn('label="4D"', graph.source)
 
+    def test_discovered_performance_dfg_retains_business_hour_slots(self):
+        import pm4py
+        from pm4py.discovery import discover_performance_dfg
+        from pm4py.objects.log.obj import Event, EventLog, Trace
+
+        slots = weekday_slots(8, 16)
+        trace = Trace(
+            [
+                Event(
+                    {
+                        "concept:name": "start",
+                        "time:timestamp": datetime(2025, 1, 7, 8),
+                    }
+                ),
+                Event(
+                    {
+                        "concept:name": "finish",
+                        "time:timestamp": datetime(2025, 1, 10, 16),
+                    }
+                ),
+            ]
+        )
+
+        dfg, start_activities, end_activities = discover_performance_dfg(
+            EventLog([trace]),
+            business_hours=True,
+            business_hour_slots=slots,
+            perf_aggregation_key="median",
+        )
+        with patch("pm4py.visualization.dfg.visualizer.view") as view:
+            pm4py.view_performance_dfg(
+                dfg,
+                start_activities,
+                end_activities,
+                aggregation_measure="median",
+            )
+        graph = view.call_args.args[0]
+
+        self.assertIsInstance(dfg, dict)
+        self.assertEqual(4 * 8 * 60 * 60, dfg[("start", "finish")])
+        self.assertEqual(tuple(slots), dfg.business_hour_slots)
+        self.assertIn('label="4D"', graph.source)
+
     def test_variant_duration_uses_configured_working_day(self):
         from pm4py.visualization.variants_duration.variants.classic import (
             _format_duration,