Coverage for django_napse/core/models/histories/history.py: 76%

1import uuid 

2from datetime import datetime, timedelta 

3 

4import pandas as pd 

5from django.db import models 

6from django.utils.timezone import get_default_timezone 

7 

8from django_napse.utils.constants import HISTORY_DATAPOINT_FIELDS 

9from django_napse.utils.errors import HistoryError 

10from django_napse.utils.findable_class import FindableClass 

11from django_napse.utils.usefull_functions import process_value_from_type 

12 

13 

14class History(FindableClass, models.Model): 

15 uuid = models.UUIDField(unique=True, editable=False, default=uuid.uuid4) 

16 

17 def __str__(self) -> str: 

18 return f"HISTORY {self.uuid}" 

19 

20 def info(self, verbose=True, beacon=""): 

21 string = "" 

22 string += f"{beacon}History {self.pk}:\n" 

23 string += f"{beacon}\t{self.uuid=}\n" 

24 string += f"{beacon}\tDataPoints:\n" 

25 df_string = str(self.to_dataframe()).replace("\n", f"\n{beacon}\t\t") 

26 string += f"{beacon}\t\t{df_string}\n" 

27 if verbose: 

28 print(string) 

29 return string 

30 

31 def to_dataframe(self): 

32 all_data_points = self.data_points.all() 

33 return pd.DataFrame([data_point.to_dict() for data_point in all_data_points]) 

34 

35 @property 

36 def owner(self): 

37 return self.find().owner 

38 

39 @classmethod 

40 def get_or_create(cls, owner): 

41 if hasattr(owner, "history"): 

42 return owner.history 

43 return cls.objects.create(owner=owner) 

44 

45 def delta(self, days: int = 30) -> float: 

46 """Return the value delta between today and {days} days ago.""" 

47 # TODO: TEST IT 

48 date = datetime.now(tz=get_default_timezone()) - timedelta(days=days) 

49 data_points = self.data_points.filter(created_at__date=date.date()) 

50 

51 while not data_points.exists(): 

52 days -= 1 

53 date = date + timedelta(days=days) 

54 data_points = self.data_points.filter(created_at__date=date.date()) 

55 if days == 0 and not data_points.exists(): 

56 return 0 

57 if data_points.exists(): 

58 break 

59 

60 return data_points 

61 

62 

63class HistoryDataPoint(models.Model): 

64 history = models.ForeignKey(History, on_delete=models.CASCADE, related_name="data_points") 

65 created_at = models.DateTimeField(auto_now_add=True) 

66 

67 def __str__(self) -> str: # pragma: no cover 

68 return f"HISTORY DATA POINT {self.pk} {self.history.uuid}" 

69 

70 def to_dict(self): 

71 return {field.key: field.get_value() for field in self.fields.all()} 

72 

73 

74class HistoryDataPointField(models.Model): 

75 history_data_point = models.ForeignKey(HistoryDataPoint, on_delete=models.CASCADE, related_name="fields") 

76 key = models.CharField(max_length=255) 

77 value = models.CharField(max_length=255) 

78 target_type = models.CharField(max_length=255) 

79 

80 def __str__(self) -> str: # pragma: no cover 

81 return f"HISTORY DATA POINT FIELD {self.pk} {self.history_data_point.pk}" 

82 

83 def save(self, *args, **kwargs): 

84 if self.key not in HISTORY_DATAPOINT_FIELDS: 

85 error_msg = f"Invalid key {self.key} for HistoryDataPointField" 

86 raise HistoryError.InvalidDataPointFieldKey(error_msg) 

87 return super().save(*args, **kwargs) 

88 

89 def get_value(self): 

90 return process_value_from_type(value=self.value, target_type=self.target_type)