controlpi_plugins.state
Provide state plugins for all kinds of systems.
- State represents a Boolean state.
- StateAlias translates to another state-like client.
- AndState combines several state-like clients by conjunction.
- OrState combines several state-like clients by disjunction.
- AndSet sets a state due to a conjunction of other state-like clients.
- OrSet sets a state due to a disjunction of other state-like clients.
All these plugins use the following conventions:
- If their state changes they send a message containing "event": "changed" and "state": NEW STATE.
- If their state is reported due to a message, but did not change they send a message containing just "state": CURRENT STATE.
- If they receive a message containing "target": NAME and "command": "get state" they report their current state.
- If State (or any other settable state using these conventions) receives a message containing "target": NAME, "command": "set state" and "new state": STATE TO SET it changes the state accordingly. If this was really a change the corresponding event is sent. If it was already in this state a report message without "event": "changed" is sent.
- StateAlias can alias any message bus client using these conventions, not just State instances. It translates all messages described here in both directions.
- AndState and OrState instances cannot be set.
- AndState and OrState can combine any message bus clients using these conventions, not just State instances. They only react to messages containing "state" information.
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State": {"plugin": "State"},
... "Test State 2": {"plugin": "State"},
... "Test State 3": {"plugin": "State"},
... "Test State 4": {"plugin": "State"},
... "Test StateAlias": {"plugin": "StateAlias",
... "alias for": "Test State 2"},
... "Test AndState": {"plugin": "AndState",
... "states": ["Test State", "Test StateAlias"]},
... "Test OrState": {"plugin": "OrState",
... "states": ["Test State", "Test StateAlias"]},
... "Test AndSet": {"plugin": "AndSet",
... "input states": ["Test State", "Test StateAlias"],
... "output state": "Test State 3"},
... "Test OrSet": {"plugin": "OrSet",
... "input states": ["Test State", "Test StateAlias"],
... "output state": "Test State 4"}},
... [{"target": "Test AndState",
... "command": "get state"},
... {"target": "Test OrState",
... "command": "get state"},
... {"target": "Test State",
... "command": "set state", "new state": True},
... {"target": "Test StateAlias",
... "command": "set state", "new state": True},
... {"target": "Test State",
... "command": "set state", "new state": False}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test AndState', 'command': 'get state'}
test(): {'sender': 'test()', 'target': 'Test OrState', 'command': 'get state'}
test(): {'sender': 'Test AndState', 'state': False}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test OrState', 'state': False}
test(): {'sender': 'test()', 'target': 'Test StateAlias',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test StateAlias', 'target': 'Test State 2',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True}
test(): {'sender': 'Test OrSet', 'target': 'Test State 4',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State', 'event': 'changed', 'state': False}
test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
test(): {'sender': 'Test State 4', 'event': 'changed', 'state': True}
1"""Provide state plugins for all kinds of systems. 2 3- State represents a Boolean state. 4- StateAlias translates to another state-like client. 5- AndState combines several state-like clients by conjunction. 6- OrState combines several state-like clients by disjunction. 7- AndSet sets a state due to a conjunction of other state-like clients. 8- OrSet sets a state due to a disjunction of other state-like clients. 9 10All these plugins use the following conventions: 11 12- If their state changes they send a message containing "event": "changed" 13 and "state": NEW STATE. 14- If their state is reported due to a message, but did not change they send 15 a message containing just "state": CURRENT STATE. 16- If they receive a message containing "target": NAME and 17 "command": "get state" they report their current state. 18- If State (or any other settable state using these conventions) receives 19 a message containing "target": NAME, "command": "set state" and 20 "new state": STATE TO SET it changes the state accordingly. If this 21 was really a change the corresponding event is sent. If it was already in 22 this state a report message without "event": "changed" is sent. 23- StateAlias can alias any message bus client using these conventions, not 24 just State instances. It translates all messages described here in both 25 directions. 26- AndState and OrState instances cannot be set. 27- AndState and OrState can combine any message bus clients using these 28 conventions, not just State instances. They only react to messages 29 containing "state" information. 30 31>>> import asyncio 32>>> import controlpi 33>>> asyncio.run(controlpi.test( 34... {"Test State": {"plugin": "State"}, 35... "Test State 2": {"plugin": "State"}, 36... "Test State 3": {"plugin": "State"}, 37... "Test State 4": {"plugin": "State"}, 38... "Test StateAlias": {"plugin": "StateAlias", 39... "alias for": "Test State 2"}, 40... "Test AndState": {"plugin": "AndState", 41... "states": ["Test State", "Test StateAlias"]}, 42... "Test OrState": {"plugin": "OrState", 43... "states": ["Test State", "Test StateAlias"]}, 44... "Test AndSet": {"plugin": "AndSet", 45... "input states": ["Test State", "Test StateAlias"], 46... "output state": "Test State 3"}, 47... "Test OrSet": {"plugin": "OrSet", 48... "input states": ["Test State", "Test StateAlias"], 49... "output state": "Test State 4"}}, 50... [{"target": "Test AndState", 51... "command": "get state"}, 52... {"target": "Test OrState", 53... "command": "get state"}, 54... {"target": "Test State", 55... "command": "set state", "new state": True}, 56... {"target": "Test StateAlias", 57... "command": "set state", "new state": True}, 58... {"target": "Test State", 59... "command": "set state", "new state": False}])) 60... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 61test(): {'sender': '', 'event': 'registered', ... 62test(): {'sender': 'test()', 'target': 'Test AndState', 'command': 'get state'} 63test(): {'sender': 'test()', 'target': 'Test OrState', 'command': 'get state'} 64test(): {'sender': 'Test AndState', 'state': False} 65test(): {'sender': 'test()', 'target': 'Test State', 66 'command': 'set state', 'new state': True} 67test(): {'sender': 'Test OrState', 'state': False} 68test(): {'sender': 'test()', 'target': 'Test StateAlias', 69 'command': 'set state', 'new state': True} 70test(): {'sender': 'Test State', 'event': 'changed', 'state': True} 71test(): {'sender': 'test()', 'target': 'Test State', 72 'command': 'set state', 'new state': False} 73test(): {'sender': 'Test StateAlias', 'target': 'Test State 2', 74 'command': 'set state', 'new state': True} 75test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True} 76test(): {'sender': 'Test OrSet', 'target': 'Test State 4', 77 'command': 'set state', 'new state': True} 78test(): {'sender': 'Test State', 'event': 'changed', 'state': False} 79test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 80test(): {'sender': 'Test State 4', 'event': 'changed', 'state': True} 81""" 82 83from controlpi import BasePlugin, Message, MessageTemplate 84 85from typing import Dict, List 86 87 88class State(BasePlugin): 89 """Provide a Boolean state. 90 91 The state of a State plugin instance can be queried with the "get state" 92 command and set with the "set state" command to the new state given by 93 the "new state" key: 94 >>> import asyncio 95 >>> import controlpi 96 >>> asyncio.run(controlpi.test( 97 ... {"Test State": {"plugin": "State"}}, 98 ... [{"target": "Test State", "command": "get state"}, 99 ... {"target": "Test State", "command": "set state", 100 ... "new state": True}, 101 ... {"target": "Test State", "command": "set state", 102 ... "new state": True}, 103 ... {"target": "Test State", "command": "get state"}])) 104 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 105 test(): {'sender': '', 'event': 'registered', ... 106 test(): {'sender': 'test()', 'target': 'Test State', 107 'command': 'get state'} 108 test(): {'sender': 'test()', 'target': 'Test State', 109 'command': 'set state', 'new state': True} 110 test(): {'sender': 'Test State', 'state': False} 111 test(): {'sender': 'test()', 'target': 'Test State', 112 'command': 'set state', 'new state': True} 113 test(): {'sender': 'Test State', 'event': 'changed', 'state': True} 114 test(): {'sender': 'test()', 'target': 'Test State', 115 'command': 'get state'} 116 test(): {'sender': 'Test State', 'state': True} 117 test(): {'sender': 'Test State', 'state': True} 118 """ 119 120 CONF_SCHEMA = True 121 """Schema for State plugin configuration. 122 123 There are no required or optional configuration keys. 124 """ 125 126 def process_conf(self) -> None: 127 """Register plugin as bus client.""" 128 self.state: bool = False 129 self.bus.register( 130 self.name, 131 "State", 132 [ 133 MessageTemplate( 134 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 135 ), 136 MessageTemplate({"state": {"type": "boolean"}}), 137 ], 138 [ 139 ( 140 [ 141 MessageTemplate( 142 { 143 "target": {"const": self.name}, 144 "command": {"const": "get state"}, 145 } 146 ) 147 ], 148 self._get_state, 149 ), 150 ( 151 [ 152 MessageTemplate( 153 { 154 "target": {"const": self.name}, 155 "command": {"const": "set state"}, 156 "new state": {"type": "boolean"}, 157 } 158 ) 159 ], 160 self._set_state, 161 ), 162 ], 163 ) 164 165 async def _get_state(self, message: Message) -> None: 166 await self.bus.send(Message(self.name, {"state": self.state})) 167 168 async def _set_state(self, message: Message) -> None: 169 if self.state != message["new state"]: 170 assert isinstance(message["new state"], bool) 171 self.state = message["new state"] 172 await self.bus.send( 173 Message(self.name, {"event": "changed", "state": self.state}) 174 ) 175 else: 176 await self.bus.send(Message(self.name, {"state": self.state})) 177 178 async def run(self) -> None: 179 """Run no code proactively.""" 180 pass 181 182 183class StateAlias(BasePlugin): 184 """Define an alias for another state. 185 186 The "alias for" configuration key gets the name for the other state that 187 is aliased by the StateAlias plugin instance. 188 189 The "get state" and "set state" commands are forwarded to and the 190 "changed" events and "state" messages are forwarded from this other 191 state: 192 >>> import asyncio 193 >>> import controlpi 194 >>> asyncio.run(controlpi.test( 195 ... {"Test State": {"plugin": "State"}, 196 ... "Test StateAlias": {"plugin": "StateAlias", 197 ... "alias for": "Test State"}}, 198 ... [{"target": "Test State", "command": "get state"}, 199 ... {"target": "Test StateAlias", "command": "set state", 200 ... "new state": True}, 201 ... {"target": "Test State", "command": "set state", 202 ... "new state": True}, 203 ... {"target": "Test StateAlias", "command": "get state"}])) 204 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 205 test(): {'sender': '', 'event': 'registered', ... 206 test(): {'sender': 'test()', 'target': 'Test State', 207 'command': 'get state'} 208 test(): {'sender': 'test()', 'target': 'Test StateAlias', 209 'command': 'set state', 'new state': True} 210 test(): {'sender': 'Test State', 'state': False} 211 test(): {'sender': 'test()', 'target': 'Test State', 212 'command': 'set state', 'new state': True} 213 test(): {'sender': 'Test StateAlias', 'target': 'Test State', 214 'command': 'set state', 'new state': True} 215 test(): {'sender': 'Test StateAlias', 'state': False} 216 test(): {'sender': 'test()', 'target': 'Test StateAlias', 217 'command': 'get state'} 218 test(): {'sender': 'Test State', 'event': 'changed', 'state': True} 219 test(): {'sender': 'Test State', 'state': True} 220 test(): {'sender': 'Test StateAlias', 'target': 'Test State', 221 'command': 'get state'} 222 test(): {'sender': 'Test StateAlias', 'event': 'changed', 'state': True} 223 test(): {'sender': 'Test StateAlias', 'state': True} 224 """ 225 226 CONF_SCHEMA = { 227 "properties": {"alias for": {"type": "string"}}, 228 "required": ["alias for"], 229 } 230 """Schema for StateAlias plugin configuration. 231 232 Required configuration key: 233 234 - 'alias for': name of aliased state. 235 """ 236 237 def process_conf(self) -> None: 238 """Register plugin as bus client.""" 239 self.bus.register( 240 self.name, 241 "StateAlias", 242 [ 243 MessageTemplate( 244 { 245 "target": {"const": self.conf["alias for"]}, 246 "command": {"const": "get state"}, 247 } 248 ), 249 MessageTemplate( 250 { 251 "target": {"const": self.conf["alias for"]}, 252 "command": {"const": "set state"}, 253 "new state": {"type": "boolean"}, 254 } 255 ), 256 MessageTemplate( 257 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 258 ), 259 MessageTemplate({"state": {"type": "boolean"}}), 260 ], 261 [ 262 ( 263 [ 264 MessageTemplate( 265 { 266 "target": {"const": self.name}, 267 "command": {"const": "get state"}, 268 } 269 ) 270 ], 271 self._get_state, 272 ), 273 ( 274 [ 275 MessageTemplate( 276 { 277 "target": {"const": self.name}, 278 "command": {"const": "set state"}, 279 "new state": {"type": "boolean"}, 280 } 281 ) 282 ], 283 self._set_state, 284 ), 285 ( 286 [ 287 MessageTemplate( 288 { 289 "sender": {"const": self.conf["alias for"]}, 290 "state": {"type": "boolean"}, 291 } 292 ) 293 ], 294 self._translate, 295 ), 296 ], 297 ) 298 299 async def _get_state(self, message: Message) -> None: 300 await self.bus.send( 301 Message( 302 self.name, {"target": self.conf["alias for"], "command": "get state"} 303 ) 304 ) 305 306 async def _set_state(self, message: Message) -> None: 307 await self.bus.send( 308 Message( 309 self.name, 310 { 311 "target": self.conf["alias for"], 312 "command": "set state", 313 "new state": message["new state"], 314 }, 315 ) 316 ) 317 318 async def _translate(self, message: Message) -> None: 319 alias_message = Message(self.name) 320 if "event" in message and message["event"] == "changed": 321 alias_message["event"] = "changed" 322 alias_message["state"] = message["state"] 323 await self.bus.send(alias_message) 324 325 async def run(self) -> None: 326 """Run no code proactively.""" 327 pass 328 329 330class AndState(BasePlugin): 331 """Define conjunction of states. 332 333 The "states" configuration key gets an array of states to be combined. 334 An AndState plugin client reacts to "get state" commands and sends 335 "changed" events when a change in one of the combined states leads to 336 a change for the conjunction: 337 >>> import asyncio 338 >>> import controlpi 339 >>> asyncio.run(controlpi.test( 340 ... {"Test State 1": {"plugin": "State"}, 341 ... "Test State 2": {"plugin": "State"}, 342 ... "Test AndState": {"plugin": "AndState", 343 ... "states": ["Test State 1", "Test State 2"]}}, 344 ... [{"target": "Test State 1", "command": "set state", 345 ... "new state": True}, 346 ... {"target": "Test State 2", "command": "set state", 347 ... "new state": True}, 348 ... {"target": "Test State 1", "command": "set state", 349 ... "new state": False}, 350 ... {"target": "Test AndState", "command": "get state"}, 351 ... {"target": "Test AndState", "command": "get sources"}])) 352 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 353 test(): {'sender': '', 'event': 'registered', ... 354 test(): {'sender': 'test()', 'target': 'Test State 1', 355 'command': 'set state', 'new state': True} 356 test(): {'sender': 'test()', 'target': 'Test State 2', 357 'command': 'set state', 'new state': True} 358 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 359 test(): {'sender': 'test()', 'target': 'Test State 1', 360 'command': 'set state', 'new state': False} 361 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 362 test(): {'sender': 'test()', 'target': 'Test AndState', 363 'command': 'get state'} 364 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 365 test(): {'sender': 'Test AndState', 'event': 'changed', 'state': True} 366 test(): {'sender': 'test()', 'target': 'Test AndState', 367 'command': 'get sources'} 368 test(): {'sender': 'Test AndState', 'state': True} 369 test(): {'sender': 'Test AndState', 'event': 'changed', 'state': False} 370 test(): {'sender': 'Test AndState', 371 'states': ['Test State 1', 'Test State 2']} 372 """ 373 374 CONF_SCHEMA = { 375 "properties": {"states": {"type": "array", "items": {"type": "string"}}}, 376 "required": ["states"], 377 } 378 """Schema for AndState plugin configuration. 379 380 Required configuration key: 381 382 - 'states': list of names of combined states. 383 """ 384 385 def process_conf(self) -> None: 386 """Register plugin as bus client.""" 387 updates: List[MessageTemplate] = [] 388 self.states: Dict[str, bool] = {} 389 for state in self.conf["states"]: 390 updates.append( 391 MessageTemplate( 392 {"sender": {"const": state}, "state": {"type": "boolean"}} 393 ) 394 ) 395 self.states[state] = False 396 self.state: bool = all(self.states.values()) 397 self.bus.register( 398 self.name, 399 "AndState", 400 [ 401 MessageTemplate( 402 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 403 ), 404 MessageTemplate({"state": {"type": "boolean"}}), 405 MessageTemplate( 406 {"states": {"type": "array", "items": {"type": "string"}}} 407 ), 408 ], 409 [ 410 ( 411 [ 412 MessageTemplate( 413 { 414 "target": {"const": self.name}, 415 "command": {"const": "get state"}, 416 } 417 ) 418 ], 419 self._get_state, 420 ), 421 ( 422 [ 423 MessageTemplate( 424 { 425 "target": {"const": self.name}, 426 "command": {"const": "get sources"}, 427 } 428 ) 429 ], 430 self._get_sources, 431 ), 432 (updates, self._update), 433 ], 434 ) 435 436 async def _get_state(self, message: Message) -> None: 437 await self.bus.send(Message(self.name, {"state": self.state})) 438 439 async def _get_sources(self, message: Message) -> None: 440 source_states = list(self.states.keys()) 441 await self.bus.send(Message(self.name, {"states": source_states})) 442 443 async def _update(self, message: Message) -> None: 444 assert isinstance(message["sender"], str) 445 assert isinstance(message["state"], bool) 446 self.states[message["sender"]] = message["state"] 447 new_state = all(self.states.values()) 448 if self.state != new_state: 449 self.state = new_state 450 await self.bus.send( 451 Message(self.name, {"event": "changed", "state": self.state}) 452 ) 453 454 async def run(self) -> None: 455 """Run no code proactively.""" 456 pass 457 458 459class OrState(BasePlugin): 460 """Define disjunction of states. 461 462 The "states" configuration key gets an array of states to be combined. 463 An OrState plugin client reacts to "get state" commands and sends 464 "changed" events when a change in one of the combined states leads to 465 a change for the disjunction: 466 >>> import asyncio 467 >>> import controlpi 468 >>> asyncio.run(controlpi.test( 469 ... {"Test State 1": {"plugin": "State"}, 470 ... "Test State 2": {"plugin": "State"}, 471 ... "Test OrState": {"plugin": "OrState", 472 ... "states": ["Test State 1", "Test State 2"]}}, 473 ... [{"target": "Test State 1", "command": "set state", 474 ... "new state": True}, 475 ... {"target": "Test State 2", "command": "set state", 476 ... "new state": True}, 477 ... {"target": "Test State 1", "command": "set state", 478 ... "new state": False}, 479 ... {"target": "Test OrState", "command": "get state"}, 480 ... {"target": "Test OrState", "command": "get sources"}])) 481 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 482 test(): {'sender': '', 'event': 'registered', ... 483 test(): {'sender': 'test()', 'target': 'Test State 1', 484 'command': 'set state', 'new state': True} 485 test(): {'sender': 'test()', 'target': 'Test State 2', 486 'command': 'set state', 'new state': True} 487 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 488 test(): {'sender': 'test()', 'target': 'Test State 1', 489 'command': 'set state', 'new state': False} 490 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 491 test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True} 492 test(): {'sender': 'test()', 'target': 'Test OrState', 493 'command': 'get state'} 494 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 495 test(): {'sender': 'test()', 'target': 'Test OrState', 496 'command': 'get sources'} 497 test(): {'sender': 'Test OrState', 'state': True} 498 test(): {'sender': 'Test OrState', 499 'states': ['Test State 1', 'Test State 2']} 500 """ 501 502 CONF_SCHEMA = { 503 "properties": {"states": {"type": "array", "items": {"type": "string"}}}, 504 "required": ["states"], 505 } 506 """Schema for OrState plugin configuration. 507 508 Required configuration key: 509 510 - 'states': list of names of combined states. 511 """ 512 513 def process_conf(self) -> None: 514 """Register plugin as bus client.""" 515 updates: List[MessageTemplate] = [] 516 self.states: Dict[str, bool] = {} 517 for state in self.conf["states"]: 518 updates.append( 519 MessageTemplate( 520 {"sender": {"const": state}, "state": {"type": "boolean"}} 521 ) 522 ) 523 self.states[state] = False 524 self.state: bool = any(self.states.values()) 525 self.bus.register( 526 self.name, 527 "OrState", 528 [ 529 MessageTemplate( 530 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 531 ), 532 MessageTemplate({"state": {"type": "boolean"}}), 533 MessageTemplate( 534 {"states": {"type": "array", "items": {"type": "string"}}} 535 ), 536 ], 537 [ 538 ( 539 [ 540 MessageTemplate( 541 { 542 "target": {"const": self.name}, 543 "command": {"const": "get state"}, 544 } 545 ) 546 ], 547 self._get_state, 548 ), 549 ( 550 [ 551 MessageTemplate( 552 { 553 "target": {"const": self.name}, 554 "command": {"const": "get sources"}, 555 } 556 ) 557 ], 558 self._get_sources, 559 ), 560 (updates, self._update), 561 ], 562 ) 563 564 async def _get_state(self, message: Message) -> None: 565 await self.bus.send(Message(self.name, {"state": self.state})) 566 567 async def _get_sources(self, message: Message) -> None: 568 source_states = list(self.states.keys()) 569 await self.bus.send(Message(self.name, {"states": source_states})) 570 571 async def _update(self, message: Message) -> None: 572 assert isinstance(message["sender"], str) 573 assert isinstance(message["state"], bool) 574 self.states[message["sender"]] = message["state"] 575 new_state = any(self.states.values()) 576 if self.state != new_state: 577 self.state = new_state 578 await self.bus.send( 579 Message(self.name, {"event": "changed", "state": self.state}) 580 ) 581 582 async def run(self) -> None: 583 """Run no code proactively.""" 584 pass 585 586 587class AndSet(BasePlugin): 588 """Set state based on conjunction of other states. 589 590 The "input states" configuration key gets an array of states used to 591 determine the state in the "output state" configuration key: 592 >>> import asyncio 593 >>> import controlpi 594 >>> asyncio.run(controlpi.test( 595 ... {"Test State 1": {"plugin": "State"}, 596 ... "Test State 2": {"plugin": "State"}, 597 ... "Test State 3": {"plugin": "State"}, 598 ... "Test AndSet": {"plugin": "AndSet", 599 ... "input states": ["Test State 1", 600 ... "Test State 2"], 601 ... "output state": "Test State 3"}}, 602 ... [{"target": "Test State 1", "command": "set state", 603 ... "new state": True}, 604 ... {"target": "Test State 2", "command": "set state", 605 ... "new state": True}, 606 ... {"target": "Test AndSet", "command": "get state"}, 607 ... {"target": "Test State 1", "command": "set state", 608 ... "new state": False}, 609 ... {"target": "Test AndSet", "command": "get state"}, 610 ... {"target": "Test AndSet", "command": "get sources"}])) 611 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 612 test(): {'sender': '', 'event': 'registered', ... 613 test(): {'sender': 'test()', 'target': 'Test State 1', 614 'command': 'set state', 'new state': True} 615 test(): {'sender': 'test()', 'target': 'Test State 2', 616 'command': 'set state', 'new state': True} 617 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 618 test(): {'sender': 'test()', 'target': 'Test AndSet', 619 'command': 'get state'} 620 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 621 test(): {'sender': 'test()', 'target': 'Test State 1', 622 'command': 'set state', 'new state': False} 623 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 624 'command': 'set state', 'new state': False} 625 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 626 'command': 'set state', 'new state': True} 627 test(): {'sender': 'test()', 'target': 'Test AndSet', 628 'command': 'get state'} 629 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 630 test(): {'sender': 'Test State 3', 'state': False} 631 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True} 632 test(): {'sender': 'test()', 'target': 'Test AndSet', 633 'command': 'get sources'} 634 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 635 'command': 'set state', 'new state': True} 636 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 637 'command': 'set state', 'new state': False} 638 test(): {'sender': 'Test AndSet', 639 'states': ['Test State 1', 'Test State 2']} 640 test(): {'sender': 'Test State 3', 'state': True} 641 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': False} 642 """ 643 644 CONF_SCHEMA = { 645 "properties": { 646 "input states": {"type": "array", "items": {"type": "string"}}, 647 "output state": {"type": "string"}, 648 }, 649 "required": ["input states", "output state"], 650 } 651 """Schema for AndSet plugin configuration. 652 653 Required configuration keys: 654 655 - 'input states': list of names of combined states. 656 - 'output state': name of state to be set. 657 """ 658 659 def process_conf(self) -> None: 660 """Register plugin as bus client.""" 661 updates: List[MessageTemplate] = [] 662 self.states: Dict[str, bool] = {} 663 for state in self.conf["input states"]: 664 updates.append( 665 MessageTemplate( 666 {"sender": {"const": state}, "state": {"type": "boolean"}} 667 ) 668 ) 669 self.states[state] = False 670 self.state: bool = all(self.states.values()) 671 self.bus.register( 672 self.name, 673 "AndSet", 674 [ 675 MessageTemplate( 676 { 677 "target": {"const": self.conf["output state"]}, 678 "command": {"const": "set state"}, 679 "new state": {"type": "boolean"}, 680 } 681 ), 682 MessageTemplate( 683 {"states": {"type": "array", "items": {"type": "string"}}} 684 ), 685 ], 686 [ 687 ( 688 [ 689 MessageTemplate( 690 { 691 "target": {"const": self.name}, 692 "command": {"const": "get state"}, 693 } 694 ) 695 ], 696 self._get_state, 697 ), 698 ( 699 [ 700 MessageTemplate( 701 { 702 "target": {"const": self.name}, 703 "command": {"const": "get sources"}, 704 } 705 ) 706 ], 707 self._get_sources, 708 ), 709 (updates, self._update), 710 ], 711 ) 712 713 async def _get_state(self, message: Message) -> None: 714 await self.bus.send( 715 Message( 716 self.name, 717 { 718 "target": self.conf["output state"], 719 "command": "set state", 720 "new state": self.state, 721 }, 722 ) 723 ) 724 725 async def _get_sources(self, message: Message) -> None: 726 source_states = list(self.states.keys()) 727 await self.bus.send(Message(self.name, {"states": source_states})) 728 729 async def _update(self, message: Message) -> None: 730 assert isinstance(message["sender"], str) 731 assert isinstance(message["state"], bool) 732 self.states[message["sender"]] = message["state"] 733 new_state = all(self.states.values()) 734 if self.state != new_state: 735 self.state = new_state 736 await self.bus.send( 737 Message( 738 self.name, 739 { 740 "target": self.conf["output state"], 741 "command": "set state", 742 "new state": self.state, 743 }, 744 ) 745 ) 746 747 async def run(self) -> None: 748 """Run no code proactively.""" 749 pass 750 751 752class OrSet(BasePlugin): 753 """Set state based on disjunction of other states. 754 755 The "input states" configuration key gets an array of states used to 756 determine the state in the "output state" configuration key: 757 >>> import asyncio 758 >>> import controlpi 759 >>> asyncio.run(controlpi.test( 760 ... {"Test State 1": {"plugin": "State"}, 761 ... "Test State 2": {"plugin": "State"}, 762 ... "Test State 3": {"plugin": "State"}, 763 ... "Test OrSet": {"plugin": "OrSet", 764 ... "input states": ["Test State 1", 765 ... "Test State 2"], 766 ... "output state": "Test State 3"}}, 767 ... [{"target": "Test State 1", "command": "set state", 768 ... "new state": True}, 769 ... {"target": "Test OrSet", "command": "get state"}, 770 ... {"target": "Test State 2", "command": "set state", 771 ... "new state": True}, 772 ... {"target": "Test State 1", "command": "set state", 773 ... "new state": False}, 774 ... {"target": "Test OrSet", "command": "get sources"}])) 775 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 776 test(): {'sender': '', 'event': 'registered', ... 777 test(): {'sender': 'test()', 'target': 'Test State 1', 778 'command': 'set state', 'new state': True} 779 test(): {'sender': 'test()', 'target': 'Test OrSet', 780 'command': 'get state'} 781 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 782 test(): {'sender': 'test()', 'target': 'Test State 2', 783 'command': 'set state', 'new state': True} 784 test(): {'sender': 'Test OrSet', 'target': 'Test State 3', 785 'command': 'set state', 'new state': False} 786 test(): {'sender': 'Test OrSet', 'target': 'Test State 3', 787 'command': 'set state', 'new state': True} 788 test(): {'sender': 'test()', 'target': 'Test State 1', 789 'command': 'set state', 'new state': False} 790 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 791 test(): {'sender': 'Test State 3', 'state': False} 792 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True} 793 test(): {'sender': 'test()', 'target': 'Test OrSet', 794 'command': 'get sources'} 795 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 796 test(): {'sender': 'Test OrSet', 797 'states': ['Test State 1', 'Test State 2']} 798 """ 799 800 CONF_SCHEMA = { 801 "properties": { 802 "input states": {"type": "array", "items": {"type": "string"}}, 803 "output state": {"type": "string"}, 804 }, 805 "required": ["input states", "output state"], 806 } 807 """Schema for OrSet plugin configuration. 808 809 Required configuration keys: 810 811 - 'input states': list of names of combined states. 812 - 'output state': name of state to be set. 813 """ 814 815 def process_conf(self) -> None: 816 """Register plugin as bus client.""" 817 updates: List[MessageTemplate] = [] 818 self.states: Dict[str, bool] = {} 819 for state in self.conf["input states"]: 820 updates.append( 821 MessageTemplate( 822 {"sender": {"const": state}, "state": {"type": "boolean"}} 823 ) 824 ) 825 self.states[state] = False 826 self.state: bool = any(self.states.values()) 827 self.bus.register( 828 self.name, 829 "AndSet", 830 [ 831 MessageTemplate( 832 { 833 "target": {"const": self.conf["output state"]}, 834 "command": {"const": "set state"}, 835 "new state": {"type": "boolean"}, 836 } 837 ), 838 MessageTemplate( 839 {"states": {"type": "array", "items": {"type": "string"}}} 840 ), 841 ], 842 [ 843 ( 844 [ 845 MessageTemplate( 846 { 847 "target": {"const": self.name}, 848 "command": {"const": "get state"}, 849 } 850 ) 851 ], 852 self._get_state, 853 ), 854 ( 855 [ 856 MessageTemplate( 857 { 858 "target": {"const": self.name}, 859 "command": {"const": "get sources"}, 860 } 861 ) 862 ], 863 self._get_sources, 864 ), 865 (updates, self._update), 866 ], 867 ) 868 869 async def _get_state(self, message: Message) -> None: 870 await self.bus.send( 871 Message( 872 self.name, 873 { 874 "target": self.conf["output state"], 875 "command": "set state", 876 "new state": self.state, 877 }, 878 ) 879 ) 880 881 async def _get_sources(self, message: Message) -> None: 882 source_states = list(self.states.keys()) 883 await self.bus.send(Message(self.name, {"states": source_states})) 884 885 async def _update(self, message: Message) -> None: 886 assert isinstance(message["sender"], str) 887 assert isinstance(message["state"], bool) 888 self.states[message["sender"]] = message["state"] 889 new_state = any(self.states.values()) 890 if self.state != new_state: 891 self.state = new_state 892 await self.bus.send( 893 Message( 894 self.name, 895 { 896 "target": self.conf["output state"], 897 "command": "set state", 898 "new state": self.state, 899 }, 900 ) 901 ) 902 903 async def run(self) -> None: 904 """Run no code proactively.""" 905 pass
89class State(BasePlugin): 90 """Provide a Boolean state. 91 92 The state of a State plugin instance can be queried with the "get state" 93 command and set with the "set state" command to the new state given by 94 the "new state" key: 95 >>> import asyncio 96 >>> import controlpi 97 >>> asyncio.run(controlpi.test( 98 ... {"Test State": {"plugin": "State"}}, 99 ... [{"target": "Test State", "command": "get state"}, 100 ... {"target": "Test State", "command": "set state", 101 ... "new state": True}, 102 ... {"target": "Test State", "command": "set state", 103 ... "new state": True}, 104 ... {"target": "Test State", "command": "get state"}])) 105 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 106 test(): {'sender': '', 'event': 'registered', ... 107 test(): {'sender': 'test()', 'target': 'Test State', 108 'command': 'get state'} 109 test(): {'sender': 'test()', 'target': 'Test State', 110 'command': 'set state', 'new state': True} 111 test(): {'sender': 'Test State', 'state': False} 112 test(): {'sender': 'test()', 'target': 'Test State', 113 'command': 'set state', 'new state': True} 114 test(): {'sender': 'Test State', 'event': 'changed', 'state': True} 115 test(): {'sender': 'test()', 'target': 'Test State', 116 'command': 'get state'} 117 test(): {'sender': 'Test State', 'state': True} 118 test(): {'sender': 'Test State', 'state': True} 119 """ 120 121 CONF_SCHEMA = True 122 """Schema for State plugin configuration. 123 124 There are no required or optional configuration keys. 125 """ 126 127 def process_conf(self) -> None: 128 """Register plugin as bus client.""" 129 self.state: bool = False 130 self.bus.register( 131 self.name, 132 "State", 133 [ 134 MessageTemplate( 135 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 136 ), 137 MessageTemplate({"state": {"type": "boolean"}}), 138 ], 139 [ 140 ( 141 [ 142 MessageTemplate( 143 { 144 "target": {"const": self.name}, 145 "command": {"const": "get state"}, 146 } 147 ) 148 ], 149 self._get_state, 150 ), 151 ( 152 [ 153 MessageTemplate( 154 { 155 "target": {"const": self.name}, 156 "command": {"const": "set state"}, 157 "new state": {"type": "boolean"}, 158 } 159 ) 160 ], 161 self._set_state, 162 ), 163 ], 164 ) 165 166 async def _get_state(self, message: Message) -> None: 167 await self.bus.send(Message(self.name, {"state": self.state})) 168 169 async def _set_state(self, message: Message) -> None: 170 if self.state != message["new state"]: 171 assert isinstance(message["new state"], bool) 172 self.state = message["new state"] 173 await self.bus.send( 174 Message(self.name, {"event": "changed", "state": self.state}) 175 ) 176 else: 177 await self.bus.send(Message(self.name, {"state": self.state})) 178 179 async def run(self) -> None: 180 """Run no code proactively.""" 181 pass
Provide a Boolean state.
The state of a State plugin instance can be queried with the "get state" command and set with the "set state" command to the new state given by the "new state" key:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State": {"plugin": "State"}},
... [{"target": "Test State", "command": "get state"},
... {"target": "Test State", "command": "set state",
... "new state": True},
... {"target": "Test State", "command": "set state",
... "new state": True},
... {"target": "Test State", "command": "get state"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'get state'}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State', 'state': False}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'get state'}
test(): {'sender': 'Test State', 'state': True}
test(): {'sender': 'Test State', 'state': True}
CONF_SCHEMA =
True
Schema for State plugin configuration.
There are no required or optional configuration keys.
def
process_conf(self) -> None:
127 def process_conf(self) -> None: 128 """Register plugin as bus client.""" 129 self.state: bool = False 130 self.bus.register( 131 self.name, 132 "State", 133 [ 134 MessageTemplate( 135 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 136 ), 137 MessageTemplate({"state": {"type": "boolean"}}), 138 ], 139 [ 140 ( 141 [ 142 MessageTemplate( 143 { 144 "target": {"const": self.name}, 145 "command": {"const": "get state"}, 146 } 147 ) 148 ], 149 self._get_state, 150 ), 151 ( 152 [ 153 MessageTemplate( 154 { 155 "target": {"const": self.name}, 156 "command": {"const": "set state"}, 157 "new state": {"type": "boolean"}, 158 } 159 ) 160 ], 161 self._set_state, 162 ), 163 ], 164 )
Register plugin as bus client.
Inherited Members
184class StateAlias(BasePlugin): 185 """Define an alias for another state. 186 187 The "alias for" configuration key gets the name for the other state that 188 is aliased by the StateAlias plugin instance. 189 190 The "get state" and "set state" commands are forwarded to and the 191 "changed" events and "state" messages are forwarded from this other 192 state: 193 >>> import asyncio 194 >>> import controlpi 195 >>> asyncio.run(controlpi.test( 196 ... {"Test State": {"plugin": "State"}, 197 ... "Test StateAlias": {"plugin": "StateAlias", 198 ... "alias for": "Test State"}}, 199 ... [{"target": "Test State", "command": "get state"}, 200 ... {"target": "Test StateAlias", "command": "set state", 201 ... "new state": True}, 202 ... {"target": "Test State", "command": "set state", 203 ... "new state": True}, 204 ... {"target": "Test StateAlias", "command": "get state"}])) 205 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 206 test(): {'sender': '', 'event': 'registered', ... 207 test(): {'sender': 'test()', 'target': 'Test State', 208 'command': 'get state'} 209 test(): {'sender': 'test()', 'target': 'Test StateAlias', 210 'command': 'set state', 'new state': True} 211 test(): {'sender': 'Test State', 'state': False} 212 test(): {'sender': 'test()', 'target': 'Test State', 213 'command': 'set state', 'new state': True} 214 test(): {'sender': 'Test StateAlias', 'target': 'Test State', 215 'command': 'set state', 'new state': True} 216 test(): {'sender': 'Test StateAlias', 'state': False} 217 test(): {'sender': 'test()', 'target': 'Test StateAlias', 218 'command': 'get state'} 219 test(): {'sender': 'Test State', 'event': 'changed', 'state': True} 220 test(): {'sender': 'Test State', 'state': True} 221 test(): {'sender': 'Test StateAlias', 'target': 'Test State', 222 'command': 'get state'} 223 test(): {'sender': 'Test StateAlias', 'event': 'changed', 'state': True} 224 test(): {'sender': 'Test StateAlias', 'state': True} 225 """ 226 227 CONF_SCHEMA = { 228 "properties": {"alias for": {"type": "string"}}, 229 "required": ["alias for"], 230 } 231 """Schema for StateAlias plugin configuration. 232 233 Required configuration key: 234 235 - 'alias for': name of aliased state. 236 """ 237 238 def process_conf(self) -> None: 239 """Register plugin as bus client.""" 240 self.bus.register( 241 self.name, 242 "StateAlias", 243 [ 244 MessageTemplate( 245 { 246 "target": {"const": self.conf["alias for"]}, 247 "command": {"const": "get state"}, 248 } 249 ), 250 MessageTemplate( 251 { 252 "target": {"const": self.conf["alias for"]}, 253 "command": {"const": "set state"}, 254 "new state": {"type": "boolean"}, 255 } 256 ), 257 MessageTemplate( 258 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 259 ), 260 MessageTemplate({"state": {"type": "boolean"}}), 261 ], 262 [ 263 ( 264 [ 265 MessageTemplate( 266 { 267 "target": {"const": self.name}, 268 "command": {"const": "get state"}, 269 } 270 ) 271 ], 272 self._get_state, 273 ), 274 ( 275 [ 276 MessageTemplate( 277 { 278 "target": {"const": self.name}, 279 "command": {"const": "set state"}, 280 "new state": {"type": "boolean"}, 281 } 282 ) 283 ], 284 self._set_state, 285 ), 286 ( 287 [ 288 MessageTemplate( 289 { 290 "sender": {"const": self.conf["alias for"]}, 291 "state": {"type": "boolean"}, 292 } 293 ) 294 ], 295 self._translate, 296 ), 297 ], 298 ) 299 300 async def _get_state(self, message: Message) -> None: 301 await self.bus.send( 302 Message( 303 self.name, {"target": self.conf["alias for"], "command": "get state"} 304 ) 305 ) 306 307 async def _set_state(self, message: Message) -> None: 308 await self.bus.send( 309 Message( 310 self.name, 311 { 312 "target": self.conf["alias for"], 313 "command": "set state", 314 "new state": message["new state"], 315 }, 316 ) 317 ) 318 319 async def _translate(self, message: Message) -> None: 320 alias_message = Message(self.name) 321 if "event" in message and message["event"] == "changed": 322 alias_message["event"] = "changed" 323 alias_message["state"] = message["state"] 324 await self.bus.send(alias_message) 325 326 async def run(self) -> None: 327 """Run no code proactively.""" 328 pass
Define an alias for another state.
The "alias for" configuration key gets the name for the other state that is aliased by the StateAlias plugin instance.
The "get state" and "set state" commands are forwarded to and the "changed" events and "state" messages are forwarded from this other state:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State": {"plugin": "State"},
... "Test StateAlias": {"plugin": "StateAlias",
... "alias for": "Test State"}},
... [{"target": "Test State", "command": "get state"},
... {"target": "Test StateAlias", "command": "set state",
... "new state": True},
... {"target": "Test State", "command": "set state",
... "new state": True},
... {"target": "Test StateAlias", "command": "get state"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'get state'}
test(): {'sender': 'test()', 'target': 'Test StateAlias',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State', 'state': False}
test(): {'sender': 'test()', 'target': 'Test State',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test StateAlias', 'target': 'Test State',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test StateAlias', 'state': False}
test(): {'sender': 'test()', 'target': 'Test StateAlias',
'command': 'get state'}
test(): {'sender': 'Test State', 'event': 'changed', 'state': True}
test(): {'sender': 'Test State', 'state': True}
test(): {'sender': 'Test StateAlias', 'target': 'Test State',
'command': 'get state'}
test(): {'sender': 'Test StateAlias', 'event': 'changed', 'state': True}
test(): {'sender': 'Test StateAlias', 'state': True}
CONF_SCHEMA =
{'properties': {'alias for': {'type': 'string'}}, 'required': ['alias for']}
Schema for StateAlias plugin configuration.
Required configuration key:
- 'alias for': name of aliased state.
def
process_conf(self) -> None:
238 def process_conf(self) -> None: 239 """Register plugin as bus client.""" 240 self.bus.register( 241 self.name, 242 "StateAlias", 243 [ 244 MessageTemplate( 245 { 246 "target": {"const": self.conf["alias for"]}, 247 "command": {"const": "get state"}, 248 } 249 ), 250 MessageTemplate( 251 { 252 "target": {"const": self.conf["alias for"]}, 253 "command": {"const": "set state"}, 254 "new state": {"type": "boolean"}, 255 } 256 ), 257 MessageTemplate( 258 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 259 ), 260 MessageTemplate({"state": {"type": "boolean"}}), 261 ], 262 [ 263 ( 264 [ 265 MessageTemplate( 266 { 267 "target": {"const": self.name}, 268 "command": {"const": "get state"}, 269 } 270 ) 271 ], 272 self._get_state, 273 ), 274 ( 275 [ 276 MessageTemplate( 277 { 278 "target": {"const": self.name}, 279 "command": {"const": "set state"}, 280 "new state": {"type": "boolean"}, 281 } 282 ) 283 ], 284 self._set_state, 285 ), 286 ( 287 [ 288 MessageTemplate( 289 { 290 "sender": {"const": self.conf["alias for"]}, 291 "state": {"type": "boolean"}, 292 } 293 ) 294 ], 295 self._translate, 296 ), 297 ], 298 )
Register plugin as bus client.
Inherited Members
331class AndState(BasePlugin): 332 """Define conjunction of states. 333 334 The "states" configuration key gets an array of states to be combined. 335 An AndState plugin client reacts to "get state" commands and sends 336 "changed" events when a change in one of the combined states leads to 337 a change for the conjunction: 338 >>> import asyncio 339 >>> import controlpi 340 >>> asyncio.run(controlpi.test( 341 ... {"Test State 1": {"plugin": "State"}, 342 ... "Test State 2": {"plugin": "State"}, 343 ... "Test AndState": {"plugin": "AndState", 344 ... "states": ["Test State 1", "Test State 2"]}}, 345 ... [{"target": "Test State 1", "command": "set state", 346 ... "new state": True}, 347 ... {"target": "Test State 2", "command": "set state", 348 ... "new state": True}, 349 ... {"target": "Test State 1", "command": "set state", 350 ... "new state": False}, 351 ... {"target": "Test AndState", "command": "get state"}, 352 ... {"target": "Test AndState", "command": "get sources"}])) 353 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 354 test(): {'sender': '', 'event': 'registered', ... 355 test(): {'sender': 'test()', 'target': 'Test State 1', 356 'command': 'set state', 'new state': True} 357 test(): {'sender': 'test()', 'target': 'Test State 2', 358 'command': 'set state', 'new state': True} 359 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 360 test(): {'sender': 'test()', 'target': 'Test State 1', 361 'command': 'set state', 'new state': False} 362 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 363 test(): {'sender': 'test()', 'target': 'Test AndState', 364 'command': 'get state'} 365 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 366 test(): {'sender': 'Test AndState', 'event': 'changed', 'state': True} 367 test(): {'sender': 'test()', 'target': 'Test AndState', 368 'command': 'get sources'} 369 test(): {'sender': 'Test AndState', 'state': True} 370 test(): {'sender': 'Test AndState', 'event': 'changed', 'state': False} 371 test(): {'sender': 'Test AndState', 372 'states': ['Test State 1', 'Test State 2']} 373 """ 374 375 CONF_SCHEMA = { 376 "properties": {"states": {"type": "array", "items": {"type": "string"}}}, 377 "required": ["states"], 378 } 379 """Schema for AndState plugin configuration. 380 381 Required configuration key: 382 383 - 'states': list of names of combined states. 384 """ 385 386 def process_conf(self) -> None: 387 """Register plugin as bus client.""" 388 updates: List[MessageTemplate] = [] 389 self.states: Dict[str, bool] = {} 390 for state in self.conf["states"]: 391 updates.append( 392 MessageTemplate( 393 {"sender": {"const": state}, "state": {"type": "boolean"}} 394 ) 395 ) 396 self.states[state] = False 397 self.state: bool = all(self.states.values()) 398 self.bus.register( 399 self.name, 400 "AndState", 401 [ 402 MessageTemplate( 403 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 404 ), 405 MessageTemplate({"state": {"type": "boolean"}}), 406 MessageTemplate( 407 {"states": {"type": "array", "items": {"type": "string"}}} 408 ), 409 ], 410 [ 411 ( 412 [ 413 MessageTemplate( 414 { 415 "target": {"const": self.name}, 416 "command": {"const": "get state"}, 417 } 418 ) 419 ], 420 self._get_state, 421 ), 422 ( 423 [ 424 MessageTemplate( 425 { 426 "target": {"const": self.name}, 427 "command": {"const": "get sources"}, 428 } 429 ) 430 ], 431 self._get_sources, 432 ), 433 (updates, self._update), 434 ], 435 ) 436 437 async def _get_state(self, message: Message) -> None: 438 await self.bus.send(Message(self.name, {"state": self.state})) 439 440 async def _get_sources(self, message: Message) -> None: 441 source_states = list(self.states.keys()) 442 await self.bus.send(Message(self.name, {"states": source_states})) 443 444 async def _update(self, message: Message) -> None: 445 assert isinstance(message["sender"], str) 446 assert isinstance(message["state"], bool) 447 self.states[message["sender"]] = message["state"] 448 new_state = all(self.states.values()) 449 if self.state != new_state: 450 self.state = new_state 451 await self.bus.send( 452 Message(self.name, {"event": "changed", "state": self.state}) 453 ) 454 455 async def run(self) -> None: 456 """Run no code proactively.""" 457 pass
Define conjunction of states.
The "states" configuration key gets an array of states to be combined. An AndState plugin client reacts to "get state" commands and sends "changed" events when a change in one of the combined states leads to a change for the conjunction:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State 1": {"plugin": "State"},
... "Test State 2": {"plugin": "State"},
... "Test AndState": {"plugin": "AndState",
... "states": ["Test State 1", "Test State 2"]}},
... [{"target": "Test State 1", "command": "set state",
... "new state": True},
... {"target": "Test State 2", "command": "set state",
... "new state": True},
... {"target": "Test State 1", "command": "set state",
... "new state": False},
... {"target": "Test AndState", "command": "get state"},
... {"target": "Test AndState", "command": "get sources"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test State 2',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test AndState',
'command': 'get state'}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False}
test(): {'sender': 'Test AndState', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test AndState',
'command': 'get sources'}
test(): {'sender': 'Test AndState', 'state': True}
test(): {'sender': 'Test AndState', 'event': 'changed', 'state': False}
test(): {'sender': 'Test AndState',
'states': ['Test State 1', 'Test State 2']}
CONF_SCHEMA =
{'properties': {'states': {'type': 'array', 'items': {'type': 'string'}}}, 'required': ['states']}
Schema for AndState plugin configuration.
Required configuration key:
- 'states': list of names of combined states.
def
process_conf(self) -> None:
386 def process_conf(self) -> None: 387 """Register plugin as bus client.""" 388 updates: List[MessageTemplate] = [] 389 self.states: Dict[str, bool] = {} 390 for state in self.conf["states"]: 391 updates.append( 392 MessageTemplate( 393 {"sender": {"const": state}, "state": {"type": "boolean"}} 394 ) 395 ) 396 self.states[state] = False 397 self.state: bool = all(self.states.values()) 398 self.bus.register( 399 self.name, 400 "AndState", 401 [ 402 MessageTemplate( 403 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 404 ), 405 MessageTemplate({"state": {"type": "boolean"}}), 406 MessageTemplate( 407 {"states": {"type": "array", "items": {"type": "string"}}} 408 ), 409 ], 410 [ 411 ( 412 [ 413 MessageTemplate( 414 { 415 "target": {"const": self.name}, 416 "command": {"const": "get state"}, 417 } 418 ) 419 ], 420 self._get_state, 421 ), 422 ( 423 [ 424 MessageTemplate( 425 { 426 "target": {"const": self.name}, 427 "command": {"const": "get sources"}, 428 } 429 ) 430 ], 431 self._get_sources, 432 ), 433 (updates, self._update), 434 ], 435 )
Register plugin as bus client.
Inherited Members
460class OrState(BasePlugin): 461 """Define disjunction of states. 462 463 The "states" configuration key gets an array of states to be combined. 464 An OrState plugin client reacts to "get state" commands and sends 465 "changed" events when a change in one of the combined states leads to 466 a change for the disjunction: 467 >>> import asyncio 468 >>> import controlpi 469 >>> asyncio.run(controlpi.test( 470 ... {"Test State 1": {"plugin": "State"}, 471 ... "Test State 2": {"plugin": "State"}, 472 ... "Test OrState": {"plugin": "OrState", 473 ... "states": ["Test State 1", "Test State 2"]}}, 474 ... [{"target": "Test State 1", "command": "set state", 475 ... "new state": True}, 476 ... {"target": "Test State 2", "command": "set state", 477 ... "new state": True}, 478 ... {"target": "Test State 1", "command": "set state", 479 ... "new state": False}, 480 ... {"target": "Test OrState", "command": "get state"}, 481 ... {"target": "Test OrState", "command": "get sources"}])) 482 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 483 test(): {'sender': '', 'event': 'registered', ... 484 test(): {'sender': 'test()', 'target': 'Test State 1', 485 'command': 'set state', 'new state': True} 486 test(): {'sender': 'test()', 'target': 'Test State 2', 487 'command': 'set state', 'new state': True} 488 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 489 test(): {'sender': 'test()', 'target': 'Test State 1', 490 'command': 'set state', 'new state': False} 491 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 492 test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True} 493 test(): {'sender': 'test()', 'target': 'Test OrState', 494 'command': 'get state'} 495 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 496 test(): {'sender': 'test()', 'target': 'Test OrState', 497 'command': 'get sources'} 498 test(): {'sender': 'Test OrState', 'state': True} 499 test(): {'sender': 'Test OrState', 500 'states': ['Test State 1', 'Test State 2']} 501 """ 502 503 CONF_SCHEMA = { 504 "properties": {"states": {"type": "array", "items": {"type": "string"}}}, 505 "required": ["states"], 506 } 507 """Schema for OrState plugin configuration. 508 509 Required configuration key: 510 511 - 'states': list of names of combined states. 512 """ 513 514 def process_conf(self) -> None: 515 """Register plugin as bus client.""" 516 updates: List[MessageTemplate] = [] 517 self.states: Dict[str, bool] = {} 518 for state in self.conf["states"]: 519 updates.append( 520 MessageTemplate( 521 {"sender": {"const": state}, "state": {"type": "boolean"}} 522 ) 523 ) 524 self.states[state] = False 525 self.state: bool = any(self.states.values()) 526 self.bus.register( 527 self.name, 528 "OrState", 529 [ 530 MessageTemplate( 531 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 532 ), 533 MessageTemplate({"state": {"type": "boolean"}}), 534 MessageTemplate( 535 {"states": {"type": "array", "items": {"type": "string"}}} 536 ), 537 ], 538 [ 539 ( 540 [ 541 MessageTemplate( 542 { 543 "target": {"const": self.name}, 544 "command": {"const": "get state"}, 545 } 546 ) 547 ], 548 self._get_state, 549 ), 550 ( 551 [ 552 MessageTemplate( 553 { 554 "target": {"const": self.name}, 555 "command": {"const": "get sources"}, 556 } 557 ) 558 ], 559 self._get_sources, 560 ), 561 (updates, self._update), 562 ], 563 ) 564 565 async def _get_state(self, message: Message) -> None: 566 await self.bus.send(Message(self.name, {"state": self.state})) 567 568 async def _get_sources(self, message: Message) -> None: 569 source_states = list(self.states.keys()) 570 await self.bus.send(Message(self.name, {"states": source_states})) 571 572 async def _update(self, message: Message) -> None: 573 assert isinstance(message["sender"], str) 574 assert isinstance(message["state"], bool) 575 self.states[message["sender"]] = message["state"] 576 new_state = any(self.states.values()) 577 if self.state != new_state: 578 self.state = new_state 579 await self.bus.send( 580 Message(self.name, {"event": "changed", "state": self.state}) 581 ) 582 583 async def run(self) -> None: 584 """Run no code proactively.""" 585 pass
Define disjunction of states.
The "states" configuration key gets an array of states to be combined. An OrState plugin client reacts to "get state" commands and sends "changed" events when a change in one of the combined states leads to a change for the disjunction:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State 1": {"plugin": "State"},
... "Test State 2": {"plugin": "State"},
... "Test OrState": {"plugin": "OrState",
... "states": ["Test State 1", "Test State 2"]}},
... [{"target": "Test State 1", "command": "set state",
... "new state": True},
... {"target": "Test State 2", "command": "set state",
... "new state": True},
... {"target": "Test State 1", "command": "set state",
... "new state": False},
... {"target": "Test OrState", "command": "get state"},
... {"target": "Test OrState", "command": "get sources"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test State 2',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test OrState',
'command': 'get state'}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False}
test(): {'sender': 'test()', 'target': 'Test OrState',
'command': 'get sources'}
test(): {'sender': 'Test OrState', 'state': True}
test(): {'sender': 'Test OrState',
'states': ['Test State 1', 'Test State 2']}
CONF_SCHEMA =
{'properties': {'states': {'type': 'array', 'items': {'type': 'string'}}}, 'required': ['states']}
Schema for OrState plugin configuration.
Required configuration key:
- 'states': list of names of combined states.
def
process_conf(self) -> None:
514 def process_conf(self) -> None: 515 """Register plugin as bus client.""" 516 updates: List[MessageTemplate] = [] 517 self.states: Dict[str, bool] = {} 518 for state in self.conf["states"]: 519 updates.append( 520 MessageTemplate( 521 {"sender": {"const": state}, "state": {"type": "boolean"}} 522 ) 523 ) 524 self.states[state] = False 525 self.state: bool = any(self.states.values()) 526 self.bus.register( 527 self.name, 528 "OrState", 529 [ 530 MessageTemplate( 531 {"event": {"const": "changed"}, "state": {"type": "boolean"}} 532 ), 533 MessageTemplate({"state": {"type": "boolean"}}), 534 MessageTemplate( 535 {"states": {"type": "array", "items": {"type": "string"}}} 536 ), 537 ], 538 [ 539 ( 540 [ 541 MessageTemplate( 542 { 543 "target": {"const": self.name}, 544 "command": {"const": "get state"}, 545 } 546 ) 547 ], 548 self._get_state, 549 ), 550 ( 551 [ 552 MessageTemplate( 553 { 554 "target": {"const": self.name}, 555 "command": {"const": "get sources"}, 556 } 557 ) 558 ], 559 self._get_sources, 560 ), 561 (updates, self._update), 562 ], 563 )
Register plugin as bus client.
Inherited Members
588class AndSet(BasePlugin): 589 """Set state based on conjunction of other states. 590 591 The "input states" configuration key gets an array of states used to 592 determine the state in the "output state" configuration key: 593 >>> import asyncio 594 >>> import controlpi 595 >>> asyncio.run(controlpi.test( 596 ... {"Test State 1": {"plugin": "State"}, 597 ... "Test State 2": {"plugin": "State"}, 598 ... "Test State 3": {"plugin": "State"}, 599 ... "Test AndSet": {"plugin": "AndSet", 600 ... "input states": ["Test State 1", 601 ... "Test State 2"], 602 ... "output state": "Test State 3"}}, 603 ... [{"target": "Test State 1", "command": "set state", 604 ... "new state": True}, 605 ... {"target": "Test State 2", "command": "set state", 606 ... "new state": True}, 607 ... {"target": "Test AndSet", "command": "get state"}, 608 ... {"target": "Test State 1", "command": "set state", 609 ... "new state": False}, 610 ... {"target": "Test AndSet", "command": "get state"}, 611 ... {"target": "Test AndSet", "command": "get sources"}])) 612 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 613 test(): {'sender': '', 'event': 'registered', ... 614 test(): {'sender': 'test()', 'target': 'Test State 1', 615 'command': 'set state', 'new state': True} 616 test(): {'sender': 'test()', 'target': 'Test State 2', 617 'command': 'set state', 'new state': True} 618 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 619 test(): {'sender': 'test()', 'target': 'Test AndSet', 620 'command': 'get state'} 621 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 622 test(): {'sender': 'test()', 'target': 'Test State 1', 623 'command': 'set state', 'new state': False} 624 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 625 'command': 'set state', 'new state': False} 626 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 627 'command': 'set state', 'new state': True} 628 test(): {'sender': 'test()', 'target': 'Test AndSet', 629 'command': 'get state'} 630 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 631 test(): {'sender': 'Test State 3', 'state': False} 632 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True} 633 test(): {'sender': 'test()', 'target': 'Test AndSet', 634 'command': 'get sources'} 635 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 636 'command': 'set state', 'new state': True} 637 test(): {'sender': 'Test AndSet', 'target': 'Test State 3', 638 'command': 'set state', 'new state': False} 639 test(): {'sender': 'Test AndSet', 640 'states': ['Test State 1', 'Test State 2']} 641 test(): {'sender': 'Test State 3', 'state': True} 642 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': False} 643 """ 644 645 CONF_SCHEMA = { 646 "properties": { 647 "input states": {"type": "array", "items": {"type": "string"}}, 648 "output state": {"type": "string"}, 649 }, 650 "required": ["input states", "output state"], 651 } 652 """Schema for AndSet plugin configuration. 653 654 Required configuration keys: 655 656 - 'input states': list of names of combined states. 657 - 'output state': name of state to be set. 658 """ 659 660 def process_conf(self) -> None: 661 """Register plugin as bus client.""" 662 updates: List[MessageTemplate] = [] 663 self.states: Dict[str, bool] = {} 664 for state in self.conf["input states"]: 665 updates.append( 666 MessageTemplate( 667 {"sender": {"const": state}, "state": {"type": "boolean"}} 668 ) 669 ) 670 self.states[state] = False 671 self.state: bool = all(self.states.values()) 672 self.bus.register( 673 self.name, 674 "AndSet", 675 [ 676 MessageTemplate( 677 { 678 "target": {"const": self.conf["output state"]}, 679 "command": {"const": "set state"}, 680 "new state": {"type": "boolean"}, 681 } 682 ), 683 MessageTemplate( 684 {"states": {"type": "array", "items": {"type": "string"}}} 685 ), 686 ], 687 [ 688 ( 689 [ 690 MessageTemplate( 691 { 692 "target": {"const": self.name}, 693 "command": {"const": "get state"}, 694 } 695 ) 696 ], 697 self._get_state, 698 ), 699 ( 700 [ 701 MessageTemplate( 702 { 703 "target": {"const": self.name}, 704 "command": {"const": "get sources"}, 705 } 706 ) 707 ], 708 self._get_sources, 709 ), 710 (updates, self._update), 711 ], 712 ) 713 714 async def _get_state(self, message: Message) -> None: 715 await self.bus.send( 716 Message( 717 self.name, 718 { 719 "target": self.conf["output state"], 720 "command": "set state", 721 "new state": self.state, 722 }, 723 ) 724 ) 725 726 async def _get_sources(self, message: Message) -> None: 727 source_states = list(self.states.keys()) 728 await self.bus.send(Message(self.name, {"states": source_states})) 729 730 async def _update(self, message: Message) -> None: 731 assert isinstance(message["sender"], str) 732 assert isinstance(message["state"], bool) 733 self.states[message["sender"]] = message["state"] 734 new_state = all(self.states.values()) 735 if self.state != new_state: 736 self.state = new_state 737 await self.bus.send( 738 Message( 739 self.name, 740 { 741 "target": self.conf["output state"], 742 "command": "set state", 743 "new state": self.state, 744 }, 745 ) 746 ) 747 748 async def run(self) -> None: 749 """Run no code proactively.""" 750 pass
Set state based on conjunction of other states.
The "input states" configuration key gets an array of states used to determine the state in the "output state" configuration key:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State 1": {"plugin": "State"},
... "Test State 2": {"plugin": "State"},
... "Test State 3": {"plugin": "State"},
... "Test AndSet": {"plugin": "AndSet",
... "input states": ["Test State 1",
... "Test State 2"],
... "output state": "Test State 3"}},
... [{"target": "Test State 1", "command": "set state",
... "new state": True},
... {"target": "Test State 2", "command": "set state",
... "new state": True},
... {"target": "Test AndSet", "command": "get state"},
... {"target": "Test State 1", "command": "set state",
... "new state": False},
... {"target": "Test AndSet", "command": "get state"},
... {"target": "Test AndSet", "command": "get sources"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test State 2',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test AndSet',
'command': 'get state'}
test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test AndSet', 'target': 'Test State 3',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test AndSet', 'target': 'Test State 3',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test AndSet',
'command': 'get state'}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False}
test(): {'sender': 'Test State 3', 'state': False}
test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test AndSet',
'command': 'get sources'}
test(): {'sender': 'Test AndSet', 'target': 'Test State 3',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test AndSet', 'target': 'Test State 3',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test AndSet',
'states': ['Test State 1', 'Test State 2']}
test(): {'sender': 'Test State 3', 'state': True}
test(): {'sender': 'Test State 3', 'event': 'changed', 'state': False}
CONF_SCHEMA =
{'properties': {'input states': {'type': 'array', 'items': {'type': 'string'}}, 'output state': {'type': 'string'}}, 'required': ['input states', 'output state']}
Schema for AndSet plugin configuration.
Required configuration keys:
- 'input states': list of names of combined states.
- 'output state': name of state to be set.
def
process_conf(self) -> None:
660 def process_conf(self) -> None: 661 """Register plugin as bus client.""" 662 updates: List[MessageTemplate] = [] 663 self.states: Dict[str, bool] = {} 664 for state in self.conf["input states"]: 665 updates.append( 666 MessageTemplate( 667 {"sender": {"const": state}, "state": {"type": "boolean"}} 668 ) 669 ) 670 self.states[state] = False 671 self.state: bool = all(self.states.values()) 672 self.bus.register( 673 self.name, 674 "AndSet", 675 [ 676 MessageTemplate( 677 { 678 "target": {"const": self.conf["output state"]}, 679 "command": {"const": "set state"}, 680 "new state": {"type": "boolean"}, 681 } 682 ), 683 MessageTemplate( 684 {"states": {"type": "array", "items": {"type": "string"}}} 685 ), 686 ], 687 [ 688 ( 689 [ 690 MessageTemplate( 691 { 692 "target": {"const": self.name}, 693 "command": {"const": "get state"}, 694 } 695 ) 696 ], 697 self._get_state, 698 ), 699 ( 700 [ 701 MessageTemplate( 702 { 703 "target": {"const": self.name}, 704 "command": {"const": "get sources"}, 705 } 706 ) 707 ], 708 self._get_sources, 709 ), 710 (updates, self._update), 711 ], 712 )
Register plugin as bus client.
Inherited Members
753class OrSet(BasePlugin): 754 """Set state based on disjunction of other states. 755 756 The "input states" configuration key gets an array of states used to 757 determine the state in the "output state" configuration key: 758 >>> import asyncio 759 >>> import controlpi 760 >>> asyncio.run(controlpi.test( 761 ... {"Test State 1": {"plugin": "State"}, 762 ... "Test State 2": {"plugin": "State"}, 763 ... "Test State 3": {"plugin": "State"}, 764 ... "Test OrSet": {"plugin": "OrSet", 765 ... "input states": ["Test State 1", 766 ... "Test State 2"], 767 ... "output state": "Test State 3"}}, 768 ... [{"target": "Test State 1", "command": "set state", 769 ... "new state": True}, 770 ... {"target": "Test OrSet", "command": "get state"}, 771 ... {"target": "Test State 2", "command": "set state", 772 ... "new state": True}, 773 ... {"target": "Test State 1", "command": "set state", 774 ... "new state": False}, 775 ... {"target": "Test OrSet", "command": "get sources"}])) 776 ... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS 777 test(): {'sender': '', 'event': 'registered', ... 778 test(): {'sender': 'test()', 'target': 'Test State 1', 779 'command': 'set state', 'new state': True} 780 test(): {'sender': 'test()', 'target': 'Test OrSet', 781 'command': 'get state'} 782 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True} 783 test(): {'sender': 'test()', 'target': 'Test State 2', 784 'command': 'set state', 'new state': True} 785 test(): {'sender': 'Test OrSet', 'target': 'Test State 3', 786 'command': 'set state', 'new state': False} 787 test(): {'sender': 'Test OrSet', 'target': 'Test State 3', 788 'command': 'set state', 'new state': True} 789 test(): {'sender': 'test()', 'target': 'Test State 1', 790 'command': 'set state', 'new state': False} 791 test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True} 792 test(): {'sender': 'Test State 3', 'state': False} 793 test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True} 794 test(): {'sender': 'test()', 'target': 'Test OrSet', 795 'command': 'get sources'} 796 test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False} 797 test(): {'sender': 'Test OrSet', 798 'states': ['Test State 1', 'Test State 2']} 799 """ 800 801 CONF_SCHEMA = { 802 "properties": { 803 "input states": {"type": "array", "items": {"type": "string"}}, 804 "output state": {"type": "string"}, 805 }, 806 "required": ["input states", "output state"], 807 } 808 """Schema for OrSet plugin configuration. 809 810 Required configuration keys: 811 812 - 'input states': list of names of combined states. 813 - 'output state': name of state to be set. 814 """ 815 816 def process_conf(self) -> None: 817 """Register plugin as bus client.""" 818 updates: List[MessageTemplate] = [] 819 self.states: Dict[str, bool] = {} 820 for state in self.conf["input states"]: 821 updates.append( 822 MessageTemplate( 823 {"sender": {"const": state}, "state": {"type": "boolean"}} 824 ) 825 ) 826 self.states[state] = False 827 self.state: bool = any(self.states.values()) 828 self.bus.register( 829 self.name, 830 "AndSet", 831 [ 832 MessageTemplate( 833 { 834 "target": {"const": self.conf["output state"]}, 835 "command": {"const": "set state"}, 836 "new state": {"type": "boolean"}, 837 } 838 ), 839 MessageTemplate( 840 {"states": {"type": "array", "items": {"type": "string"}}} 841 ), 842 ], 843 [ 844 ( 845 [ 846 MessageTemplate( 847 { 848 "target": {"const": self.name}, 849 "command": {"const": "get state"}, 850 } 851 ) 852 ], 853 self._get_state, 854 ), 855 ( 856 [ 857 MessageTemplate( 858 { 859 "target": {"const": self.name}, 860 "command": {"const": "get sources"}, 861 } 862 ) 863 ], 864 self._get_sources, 865 ), 866 (updates, self._update), 867 ], 868 ) 869 870 async def _get_state(self, message: Message) -> None: 871 await self.bus.send( 872 Message( 873 self.name, 874 { 875 "target": self.conf["output state"], 876 "command": "set state", 877 "new state": self.state, 878 }, 879 ) 880 ) 881 882 async def _get_sources(self, message: Message) -> None: 883 source_states = list(self.states.keys()) 884 await self.bus.send(Message(self.name, {"states": source_states})) 885 886 async def _update(self, message: Message) -> None: 887 assert isinstance(message["sender"], str) 888 assert isinstance(message["state"], bool) 889 self.states[message["sender"]] = message["state"] 890 new_state = any(self.states.values()) 891 if self.state != new_state: 892 self.state = new_state 893 await self.bus.send( 894 Message( 895 self.name, 896 { 897 "target": self.conf["output state"], 898 "command": "set state", 899 "new state": self.state, 900 }, 901 ) 902 ) 903 904 async def run(self) -> None: 905 """Run no code proactively.""" 906 pass
Set state based on disjunction of other states.
The "input states" configuration key gets an array of states used to determine the state in the "output state" configuration key:
>>> import asyncio
>>> import controlpi
>>> asyncio.run(controlpi.test(
... {"Test State 1": {"plugin": "State"},
... "Test State 2": {"plugin": "State"},
... "Test State 3": {"plugin": "State"},
... "Test OrSet": {"plugin": "OrSet",
... "input states": ["Test State 1",
... "Test State 2"],
... "output state": "Test State 3"}},
... [{"target": "Test State 1", "command": "set state",
... "new state": True},
... {"target": "Test OrSet", "command": "get state"},
... {"target": "Test State 2", "command": "set state",
... "new state": True},
... {"target": "Test State 1", "command": "set state",
... "new state": False},
... {"target": "Test OrSet", "command": "get sources"}]))
... # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
test(): {'sender': '', 'event': 'registered', ...
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test OrSet',
'command': 'get state'}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test State 2',
'command': 'set state', 'new state': True}
test(): {'sender': 'Test OrSet', 'target': 'Test State 3',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test OrSet', 'target': 'Test State 3',
'command': 'set state', 'new state': True}
test(): {'sender': 'test()', 'target': 'Test State 1',
'command': 'set state', 'new state': False}
test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
test(): {'sender': 'Test State 3', 'state': False}
test(): {'sender': 'Test State 3', 'event': 'changed', 'state': True}
test(): {'sender': 'test()', 'target': 'Test OrSet',
'command': 'get sources'}
test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False}
test(): {'sender': 'Test OrSet',
'states': ['Test State 1', 'Test State 2']}
CONF_SCHEMA =
{'properties': {'input states': {'type': 'array', 'items': {'type': 'string'}}, 'output state': {'type': 'string'}}, 'required': ['input states', 'output state']}
Schema for OrSet plugin configuration.
Required configuration keys:
- 'input states': list of names of combined states.
- 'output state': name of state to be set.
def
process_conf(self) -> None:
816 def process_conf(self) -> None: 817 """Register plugin as bus client.""" 818 updates: List[MessageTemplate] = [] 819 self.states: Dict[str, bool] = {} 820 for state in self.conf["input states"]: 821 updates.append( 822 MessageTemplate( 823 {"sender": {"const": state}, "state": {"type": "boolean"}} 824 ) 825 ) 826 self.states[state] = False 827 self.state: bool = any(self.states.values()) 828 self.bus.register( 829 self.name, 830 "AndSet", 831 [ 832 MessageTemplate( 833 { 834 "target": {"const": self.conf["output state"]}, 835 "command": {"const": "set state"}, 836 "new state": {"type": "boolean"}, 837 } 838 ), 839 MessageTemplate( 840 {"states": {"type": "array", "items": {"type": "string"}}} 841 ), 842 ], 843 [ 844 ( 845 [ 846 MessageTemplate( 847 { 848 "target": {"const": self.name}, 849 "command": {"const": "get state"}, 850 } 851 ) 852 ], 853 self._get_state, 854 ), 855 ( 856 [ 857 MessageTemplate( 858 { 859 "target": {"const": self.name}, 860 "command": {"const": "get sources"}, 861 } 862 ) 863 ], 864 self._get_sources, 865 ), 866 (updates, self._update), 867 ], 868 )
Register plugin as bus client.