33import dataclasses
44import json
55from pathlib import Path
6- from typing import TYPE_CHECKING , Any , Generic , TypeVar
6+ from typing import TYPE_CHECKING , Any , Generic , TypeVar , cast
77
88from typing_extensions import Self
99
1010if TYPE_CHECKING :
1111 from collections .abc import Mapping
1212
13- from dingz .rest import ConfigDump , Device , NetworkInfo , Ram , State
13+ from dingz .rest import (
14+ ActionsConfig ,
15+ ButtonsConfig ,
16+ ConfigDump ,
17+ DdiConfig ,
18+ Device ,
19+ InputsConfig ,
20+ LuxConfig ,
21+ NetworkInfo ,
22+ OutputsConfig ,
23+ PirsConfig ,
24+ Ram ,
25+ SchedulerConfig ,
26+ ServicesConfig ,
27+ State ,
28+ SystemConfig ,
29+ )
1430
1531__all__ = [
32+ "REDACTED_VALUE" ,
1633 "Endpoint" ,
1734 "Error" ,
1835 "Snapshot" ,
1936]
2037
38+ REDACTED_VALUE = "[redacted]"
39+
40+
2141_TESTS_DIR = Path (__file__ ).parent
2242_SNAPSHOTS_DIR = _TESTS_DIR / "snapshots"
2343
@@ -60,39 +80,63 @@ class Snapshot:
6080 config_dump : Endpoint [ConfigDump ]
6181 state : Endpoint [State ]
6282 ram : Endpoint [Ram ]
83+ button_config : Endpoint [ButtonsConfig ]
84+ input_config : Endpoint [InputsConfig ]
85+ pir_config : Endpoint [PirsConfig ]
86+ lux_config : Endpoint [LuxConfig ]
87+ output_config : Endpoint [OutputsConfig ]
88+ services_config : Endpoint [ServicesConfig ]
89+ system_config : Endpoint [SystemConfig ]
90+ ddi_config : Endpoint [DdiConfig ]
91+ actions_config : Endpoint [ActionsConfig ]
92+ scheduler_config : Endpoint [list [SchedulerConfig ]]
6393
6494 def endpoint_by_path (self , path : str ) -> Endpoint [Any ]:
65- if path == "/api/v1/device" :
66- return self .device
67- if path == "/api/v1/info" :
68- return self .network_info
69- if path == "/api/v1/dump_config" :
70- return self .config_dump
71- if path == "/api/v1/state" :
72- return self .state
73- if path == "/api/v1/ram" :
74- return self .ram
75- msg = f"Unknown endpoint path: { path }
76- raise LookupError (msg )
95+ attr = _ENDPOINT_ATTR_MAP .get (path )
96+ if not attr :
97+ msg = f"Unknown endpoint path: { path }
98+ raise LookupError (msg )
99+ return getattr (self , attr )
77100
78101 def path (self ) -> Path :
79102 return _SNAPSHOTS_DIR / f"{ self .device_hash }
80103
81- def save (self ) -> None :
82- text = json .dumps (dataclasses .asdict (self ), indent = 2 )
83- self .path ().write_text (text , encoding = "utf-8" )
104+ def save (self , * , force_update : bool = False ) -> None :
105+ data = dataclasses .asdict (self )
106+ path = self .path ()
107+ if not force_update and path .exists ():
108+ old_data = json .loads (path .read_text (encoding = "utf-8" ))
109+ data = _preserve_old_value (old_data , data )
110+
111+ text = json .dumps (data , indent = 2 )
112+ path .write_text (text , encoding = "utf-8" )
84113
85114 @classmethod
86115 def from_dict (cls , data : Mapping [str , Any ]) -> Self :
87- return cls (
88- device_hash = data ["device_hash" ],
89- firmware_version = Endpoint .from_dict (data ["firmware_version" ]),
90- device = Endpoint .from_dict (data ["device" ]),
91- network_info = Endpoint .from_dict (data ["network_info" ]),
92- config_dump = Endpoint .from_dict (data ["config_dump" ]),
93- state = Endpoint .from_dict (data ["state" ]),
94- ram = Endpoint .from_dict (data ["ram" ]),
95- )
116+ kwargs : dict [str , Endpoint [Any ]] = {}
117+ for attr in _ENDPOINT_ATTR_MAP .values ():
118+ kwargs [attr ] = Endpoint .from_dict (data [attr ])
119+ return cls (device_hash = data ["device_hash" ], ** kwargs )
120+
121+
122+ _ENDPOINT_ATTR_MAP : dict [str , str ] = {
123+ "/api/v1/firmware" : "firmware_version" ,
124+ "/api/v1/device" : "device" ,
125+ "/api/v1/info" : "network_info" ,
126+ "/api/v1/dump_config" : "config_dump" ,
127+ "/api/v1/state" : "state" ,
128+ "/api/v1/ram" : "ram" ,
129+ "/api/v1/button_config" : "button_config" ,
130+ "/api/v1/input_config" : "input_config" ,
131+ "/api/v1/pir_config" : "pir_config" ,
132+ "/api/v1/lux_config" : "lux_config" ,
133+ "/api/v1/output_config" : "output_config" ,
134+ "/api/v1/services_config" : "services_config" ,
135+ "/api/v1/system_config" : "system_config" ,
136+ "/api/v1/ddi_config" : "ddi_config" ,
137+ "/api/v1/action" : "actions_config" ,
138+ "/api/v1/scheduler" : "scheduler_config" ,
139+ }
96140
97141
98142def load_snapshots () -> list [Snapshot ]:
@@ -101,3 +145,36 @@ def load_snapshots() -> list[Snapshot]:
101145 data = json .loads (path .read_text (encoding = "utf-8" ))
102146 snapshots .append (Snapshot .from_dict (data ))
103147 return snapshots
148+
149+
150+ def _preserve_old_value (old : Any , new : Any ) -> Any : # noqa: PLR0911
151+ """Recursively preserve old values if they haven't changed type or meaning."""
152+ if isinstance (old , float ) and isinstance (new , int ):
153+ # Value changed from float to int. The API returns ints if the value happens to align.
154+ return old
155+ if type (old ) is not type (new ):
156+ return new
157+ if isinstance (old , dict ):
158+ old = cast ("dict[Any, Any]" , old )
159+ new_dict = new .copy ()
160+ for key , old_value in old .items ():
161+ if key not in new_dict :
162+ continue
163+ new_value = new_dict [key ]
164+ new_dict [key ] = _preserve_old_value (old_value , new_value )
165+ return new_dict
166+ if isinstance (old , list ):
167+ old = cast ("list[Any]" , old )
168+ if len (old ) != len (new ):
169+ return new
170+ return [_preserve_old_value (old_item , new_item ) for old_item , new_item in zip (old , new )]
171+
172+ if new == REDACTED_VALUE :
173+ # Always use the redacted value even if old wasn't yet
174+ return REDACTED_VALUE
175+
176+ if (new == "" ) != (old == "" ):
177+ # We changed from empty to non-empty string or vice versa
178+ return new
179+
180+ return old
0 commit comments