Coverage for core / src / sensorkit / core / delegate.py: 74%

1# SPDX-License-Identifier: Apache-2.0 

2from __future__ import annotations 

3 

4from abc import abstractmethod 

5from typing import TYPE_CHECKING, Any, Callable, override 

6 

7from pydantic import BaseModel 

8 

9from sensorkit.core.controller import ( 

10 ControllerDevice, 

11 ControllerInterface, 

12 InternalControllerState, 

13) 

14from sensorkit.core.device import DeviceCommand, DeviceInterface 

15from sensorkit.core.entity import EntityInterface 

16from sensorkit.core.executor import TaskFactoryFunc 

17from sensorkit.core.program import ProgramInterface 

18from sensorkit.core.task import Task 

19 

20if TYPE_CHECKING: 

21 from collections.abc import Collection, Coroutine 

22 from datetime import datetime 

23 

24 from sensorkit.backend.event import Event 

25 from sensorkit.backend.request import ExtendedHandlerFunc, HandlerFunc, Request 

26 from sensorkit.common.keyword import Keyword 

27 from sensorkit.core.client import SensorKit 

28 from sensorkit.data.context import Context 

29 from sensorkit.data.graph import DataGraph 

30 

31 

32class EntityDelegate(EntityInterface): 

33 """Delegates EntityInterface calls to a target implementation.""" 

34 

35 @property 

36 @abstractmethod 

37 def delegate_target(self) -> EntityInterface: 

38 """Return the EntityInterface implementation to which calls are delegated.""" 

39 ... 

40 

41 @override 

42 async def kv_put_model(self, model: BaseModel): 

43 return await self.delegate_target.kv_put_model(model) 

44 

45 @override 

46 async def kv_get_model[M: BaseModel](self, model_type: type[M]) -> M: 

47 return await self.delegate_target.kv_get_model(model_type) 

48 

49 @override 

50 @property 

51 def task_group(self): 

52 return self.delegate_target.task_group 

53 

54 @override 

55 async def emit_event(self, event: Event): 

56 return await self.delegate_target.emit_event(event) 

57 

58 @override 

59 async def publish(self, model: Keyword): 

60 return await self.delegate_target.publish(model) 

61 

62 @override 

63 async def handle_request[P: BaseModel | None, R: BaseModel | None, V: BaseModel | None]( 

64 self, 

65 request: Request[P, R, V], 

66 func: HandlerFunc[P, R] | ExtendedHandlerFunc[P, R, V], 

67 ): 

68 return await self.delegate_target.handle_request(request, func) 

69 

70 @override 

71 async def data_graph(self) -> DataGraph: 

72 return await self.delegate_target.data_graph() 

73 

74 @override 

75 async def publish_entity_info(self): 

76 return await self.delegate_target.publish_entity_info() 

77 

78 @override 

79 def sensorkit(self) -> SensorKit: 

80 return self.delegate_target.sensorkit() 

81 

82 

83class ControllerDelegate(EntityDelegate, ControllerInterface): 

84 """Delegates ControllerInterface calls to a target implementation.""" 

85 

86 @property 

87 @abstractmethod 

88 def delegate_target(self) -> ControllerInterface: 

89 """Return the ControllerInterface implementation to which calls are delegated.""" 

90 ... 

91 

92 @override 

93 def on_enable(self, func: Callable[[], None]): 

94 return self.delegate_target.on_enable(func) 

95 

96 @override 

97 def on_disable(self, func: Callable[[], None]): 

98 return self.delegate_target.on_disable(func) 

99 

100 @override 

101 def use_device(self, name: str, *, subscribe: list[type] | None = None): 

102 return self.delegate_target.use_device(name, subscribe=subscribe) 

103 

104 @override 

105 def get_device(self, name: str) -> ControllerDevice: 

106 return self.delegate_target.get_device(name) 

107 

108 @override 

109 def all_devices(self) -> Collection[ControllerDevice]: 

110 return self.delegate_target.all_devices() 

111 

112 @override 

113 async def start_device_subscriptions(self): 

114 return await self.delegate_target.start_device_subscriptions() 

115 

116 @override 

117 async def stop_device_subscriptions(self): 

118 return await self.delegate_target.stop_device_subscriptions() 

119 

120 @override 

121 async def update_context(self, *args, **kwargs) -> Context: 

122 return await self.delegate_target.update_context(*args, **kwargs) 

123 

124 @override 

125 def task_handler( 

126 self, task_type: type[Task] 

127 ) -> Callable[[Callable[..., Coroutine[Any, Any, None]]], Callable[..., Coroutine[Any, Any, None]]]: 

128 return self.delegate_target.task_handler(task_type) 

129 

130 @override 

131 def task_running(self) -> bool: 

132 return self.delegate_target.task_running() 

133 

134 @override 

135 async def set_internal_state(self, state: InternalControllerState): 

136 return await self.delegate_target.set_internal_state(state) 

137 

138 

139class DeviceDelegate(EntityDelegate, DeviceInterface): 

140 """Delegates DeviceInterface calls to a target implementation.""" 

141 

142 @property 

143 @abstractmethod 

144 def delegate_target(self) -> DeviceInterface: 

145 """Return the DeviceInterface implementation to which calls are delegated.""" 

146 ... 

147 

148 @override 

149 def on_enable(self, func: Callable[[], None]): 

150 return self.delegate_target.on_enable(func) 

151 

152 @override 

153 def on_disable(self, func: Callable[[], None]): 

154 return self.delegate_target.on_disable(func) 

155 

156 @override 

157 def command_handler( 

158 self, command_type: type[DeviceCommand] 

159 ) -> Callable[..., Callable[[DeviceCommand], Coroutine[Any, Any, BaseModel | None]]]: 

160 return self.delegate_target.command_handler(command_type) 

161 

162 

163class ProgramDelegate(EntityDelegate, ProgramInterface): 

164 """Delegates ProgramInterface calls to a target implementation.""" 

165 

166 @property 

167 @abstractmethod 

168 def delegate_target(self) -> ProgramInterface: 

169 """Return the ProgramInterface implementation to which calls are delegated.""" 

170 ... 

171 

172 @override 

173 def on_enable(self, func: Callable[[], None]): 

174 return self.delegate_target.on_enable(func) 

175 

176 @override 

177 def on_disable(self, func: Callable[[], None]): 

178 return self.delegate_target.on_disable(func) 

179 

180 @override 

181 def task_factory(self, func: TaskFactoryFunc) -> TaskFactoryFunc: 

182 return self.delegate_target.task_factory(func) 

183 

184 @override 

185 def get_offers(self) -> list[Any]: 

186 return self.delegate_target.get_offers() 

187 

188 @override 

189 async def publish_offers(self): 

190 return await self.delegate_target.publish_offers() 

191 

192 @override 

193 def add_offer(self, start: datetime, end: datetime, obj: Any = None): 

194 return self.delegate_target.add_offer(start, end, obj) 

195 

196 @override 

197 def remove_offer(self, start: datetime, end: datetime, obj: Any = None): 

198 return self.delegate_target.remove_offer(start, end, obj) 

199 

200 @override 

201 def clear_offers(self): 

202 return self.delegate_target.clear_offers()