1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
"""Source module."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from enum import StrEnum
from typing import Any
from asusrouter.const import UNKNOWN_MEMBER_STR
from asusrouter.tools.enum import FromStrMixin
from asusrouter.tools.types import ARCallbackType
class ARDataSource:
"""AsusRouter data source.
This is a universal class representing a data source
within the AsusRouter ecosystem.
"""
def __init__(self) -> None:
"""Initialize the data source."""
class ARDataType(FromStrMixin, StrEnum):
"""AsusRouter static data types."""
UNKNOWN = UNKNOWN_MEMBER_STR
class ARDataState:
"""AsusRouter data state class.
Instances of this class hold information on the data
and time of the last update.
"""
def __init__(self, source: ARDataSource | ARDataType) -> None:
"""Initialize the data state."""
if not isinstance(source, ARDataSource | ARDataType):
raise TypeError(
"A valid `ARDataSource` or `ARDataType` is required "
"to initialize an <ARDataState>. "
f"Received: {type(source)}."
)
self._source: ARDataSource | ARDataType = source
self._content: Any | None = None
self._last_update: datetime | None = None
self._callback: ARCallbackType | None = None
def update(self, content: Any) -> None:
"""Update the last update timestamp to the current time."""
self._content = content
self._last_update = datetime.now(UTC)
def is_fresh(self, threshold: timedelta) -> bool:
"""Check if the data is fresh based on the given threshold."""
if not isinstance(threshold, timedelta):
raise TypeError(
"A valid `timedelta` is required to define if the data "
"is fresh."
)
if self._last_update is None:
return False
return self._last_update + threshold > datetime.now(UTC)
@property
def content(self) -> Any | None:
"""Get the content."""
return self._content
@property
def last_update(self) -> datetime | None:
"""Get the last update timestamp."""
return self._last_update
@property
def callback(self) -> ARCallbackType | None:
"""Get the callback function."""
return self._callback
@callback.setter
def callback(self, value: ARCallbackType | None) -> None:
"""Set the callback function."""
self._callback = value
class ARDataStateStatic(ARDataState):
"""AsusRouter static data state class.
This class represents a static data, meaning it can be directly
fetched without providing any additional context.
"""
def __init__(self, source: ARDataType) -> None:
"""Initialize the static data state."""
super().__init__(source)
@property
def source(self) -> ARDataType:
"""Get the static data source."""
return ARDataType.from_value(self._source)
class ARDataStateDynamic(ARDataState):
"""AsusRouter dynamic data state class.
This class represents a dynamic data state that can change
over time and requires context to be fetched.
"""
def __init__(self, source: ARDataSource) -> None:
"""Initialize the dynamic data state."""
super().__init__(source)
@property
def source(self) -> ARDataSource:
"""Get the dynamic data source."""
return (
self._source
if isinstance(self._source, ARDataSource)
else ARDataSource()
)
|
