Coverage for functions \ flipdare \ analysis \ data \ nested \ time_series_payment_data.py: 79%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12from dataclasses import dataclass 

13from typing import override 

14from datetime import datetime 

15from flipdare.analysis.data._time_series_protocol import TimeSeriesPlotInfo 

16from flipdare.analysis.data.nested._time_series_nested_data import TimeSeriesNestedData 

17from flipdare.app_types import AnalysisDataType, ReportListType 

18from flipdare.generated.shared.payment.payment_status import PaymentStatus 

19from flipdare.util.time_util import TimeUtil 

20 

21__all__ = ["PaymentStat", "TimeSeriesPaymentData"] 

22 

23 

24@dataclass(frozen=True, slots=True) 

25class PaymentStat: 

26 count: float 

27 error_count: float 

28 

29 def accumulate(self, other: "PaymentStat") -> "PaymentStat": 

30 return PaymentStat(self.count + other.count, self.error_count + other.error_count) 

31 

32 @property 

33 def is_empty(self) -> bool: 

34 return self.count == 0.0 and self.error_count == 0.0 

35 

36 

37_EMPTY_STAT = PaymentStat(0.0, 0.0) 

38 

39 

40@dataclass 

41class TimeSeriesPaymentData(TimeSeriesNestedData[PaymentStatus, PaymentStat]): 

42 @property 

43 @override 

44 def headers(self) -> list[str]: 

45 return ["Date", "Type", "Count", "ErrorCount"] 

46 

47 def add(self, dt: datetime, status: PaymentStatus, stat: PaymentStat) -> None: 

48 day = self._to_day(dt) 

49 existing = self._rows[day].get(status, _EMPTY_STAT) 

50 self._rows[day][status] = existing.accumulate(stat) 

51 

52 def merge(self, other: "TimeSeriesPaymentData") -> None: 

53 for dt, row in other._rows.items(): 

54 for status, stat in row.items(): 

55 self.add(dt, status, stat) 

56 

57 @override 

58 def table_data(self) -> ReportListType: 

59 result: ReportListType = [] 

60 for dt in self.dates: 

61 date_str = TimeUtil.formatted_short(dt) 

62 for status, stat in sorted(self._rows[dt].items(), key=lambda x: str(x[0])): 

63 result.append([date_str, status.label, stat.count, stat.error_count]) 

64 return result 

65 

66 @override 

67 def plot_info(self) -> list[TimeSeriesPlotInfo]: 

68 """One plot per charge stat type.""" 

69 dates = self.dates 

70 date_labels = [TimeUtil.formatted_short(dt) for dt in dates] 

71 statuses: list[PaymentStatus] = sorted( 

72 {col for row in self._rows.values() for col in row}, 

73 key=str, 

74 ) 

75 

76 result: list[TimeSeriesPlotInfo] = [] 

77 for status in statuses: 

78 stats = [self._rows[dt].get(status, _EMPTY_STAT) for dt in dates] 

79 data: AnalysisDataType = [ 

80 [s.count for s in stats], 

81 [s.error_count for s in stats], 

82 ] 

83 legend_labels = [f"{status.label} Count", f"{status.label} Error Count"] 

84 result.append( 

85 TimeSeriesPlotInfo( 

86 label=status.label, 

87 x_title="Date", 

88 y_title="Count", 

89 x_labels=date_labels, 

90 legend_labels=legend_labels, 

91 data=data, 

92 ) 

93 ) 

94 return result