"""Provide state plugins for all kinds of systems.
-TODO: documentation, doctests
-TODO: AndState, OrState
+- State represents a Boolean state.
+- StateAlias translates to another state-like client.
+- AndState combines several state clients by conjunction.
+- OrState combines several state clients by disjunction.
+
+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 be 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.
+- 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 use the "get state"
+ command (to initialise their own internal state) and react to messages
+ containing "state" information.
+
+>>> import asyncio
+>>> import controlpi
+>>> asyncio.run(controlpi.test(
+... {"Test State": {"plugin": "State"},
+... "Test State 2": {"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"]}},
+... [{"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
+test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State 2', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State 2'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 2'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test StateAlias', 'plugin': 'StateAlias',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}},
+ {'target': {'const': 'Test State 2'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 2'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test StateAlias'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test StateAlias'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State 2'},
+ 'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State 2'},
+ 'state': {'type': 'boolean'}}]}
+test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test AndState', 'plugin': 'AndState',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test AndState'},
+ 'command': {'const': 'get state'}},
+ {'sender': {'const': 'Test State'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test StateAlias'},
+ 'state': {'type': 'boolean'}}]}
+test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test OrState', 'plugin': 'OrState',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test OrState'},
+ 'command': {'const': 'get state'}},
+ {'sender': {'const': 'Test State'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test StateAlias'},
+ 'state': {'type': 'boolean'}}]}
+test(): {'sender': 'test()', 'target': 'Test AndState',
+ 'command': 'get state'}
+test(): {'sender': 'Test AndState', 'state': False}
+test(): {'sender': 'test()', 'target': 'Test OrState',
+ 'command': 'get state'}
+test(): {'sender': 'Test OrState', '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 OrState', 'event': 'changed', 'state': True}
+test(): {'sender': 'test()', 'target': 'Test StateAlias',
+ 'command': 'set state', 'new state': True}
+test(): {'sender': 'Test StateAlias', 'target': 'Test State 2',
+ 'command': 'set state', 'new state': True}
+test(): {'sender': 'Test State 2', 'event': 'changed', 'state': True}
+test(): {'sender': 'Test StateAlias', 'event': 'changed', 'state': True}
+test(): {'sender': 'Test AndState', 'event': 'changed', 'state': True}
+test(): {'sender': 'test()', 'target': 'Test State',
+ 'command': 'set state', 'new state': False}
+test(): {'sender': 'Test State', 'event': 'changed', 'state': False}
+test(): {'sender': 'Test AndState', 'event': 'changed', 'state': False}
"""
-from controlpi import BasePlugin, Message
+from controlpi import BasePlugin, Message, MessageTemplate
+
+from typing import Dict
class State(BasePlugin):
+ """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
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': 'test()', 'target': 'Test State',
+ 'command': 'get state'}
+ 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': 'set state', 'new state': True}
+ test(): {'sender': 'Test State', 'state': True}
+ test(): {'sender': 'test()', 'target': 'Test State',
+ 'command': 'get state'}
+ test(): {'sender': 'Test State', 'state': True}
+ """
+
CONF_SCHEMA = True
+ """Schema for State plugin configuration.
+
+ There are no required or optional configuration keys.
+ """
async def receive(self, message: Message) -> None:
+ """Process commands to get/set state."""
if message['command'] == 'get state':
- answer = {'sender': self.name, 'state': self.state}
- await self.bus.send(answer)
+ await self.bus.send(Message(self.name, {'state': self.state}))
elif message['command'] == 'set state':
if self.state != message['new state']:
+ assert isinstance(message['new state'], bool)
self.state: bool = message['new state']
- event = {'sender': self.name, 'event': 'changed',
- 'state': self.state}
- await self.bus.send(event)
+ await self.bus.send(Message(self.name,
+ {'event': 'changed',
+ 'state': self.state}))
else:
- answer = {'sender': self.name, 'state': self.state}
- await self.bus.send(answer)
+ await self.bus.send(Message(self.name,
+ {'state': self.state}))
def process_conf(self) -> None:
+ """Register plugin as bus client."""
self.state = False
- sends = [{'event': {'const': 'changed'},
- 'state': {'type': 'boolean'}},
- {'state': {'type': 'boolean'}}]
- receives = [{'target': {'const': self.name},
- 'command': {'const': 'get state'}},
- {'target': {'const': self.name},
- 'command': {'const': 'set state'},
- 'new state': {'type': 'boolean'}}]
- self.bus.register(self.name, 'State',
- sends,
- receives,
- self.receive)
+ sends = [MessageTemplate({'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}}),
+ MessageTemplate({'state': {'type': 'boolean'}})]
+ receives = [MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'get state'}}),
+ MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}})]
+ self.bus.register(self.name, 'State', sends, receives, self.receive)
+
+ async def run(self) -> None:
+ """Run no code proactively."""
+ pass
+
+
+class StateAlias(BasePlugin):
+ """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
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test StateAlias', 'plugin': 'StateAlias',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}},
+ {'target': {'const': 'Test State'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test StateAlias'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test StateAlias'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State'},
+ 'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State'},
+ 'state': {'type': 'boolean'}}]}
+ test(): {'sender': 'test()', 'target': 'Test State',
+ 'command': 'get state'}
+ test(): {'sender': 'Test State', 'state': False}
+ test(): {'sender': 'Test StateAlias', 'state': False}
+ test(): {'sender': 'test()', 'target': 'Test StateAlias',
+ 'command': 'set state', 'new state': True}
+ test(): {'sender': 'Test StateAlias', 'target': 'Test State',
+ 'command': 'set state', 'new state': True}
+ test(): {'sender': 'Test State', 'event': 'changed', 'state': True}
+ test(): {'sender': 'Test StateAlias', 'event': 'changed', 'state': True}
+ test(): {'sender': 'test()', 'target': 'Test State',
+ 'command': 'set state', 'new state': True}
+ test(): {'sender': 'Test State', 'state': True}
+ test(): {'sender': 'Test StateAlias', 'state': True}
+ test(): {'sender': 'test()', 'target': 'Test StateAlias',
+ 'command': 'get state'}
+ test(): {'sender': 'Test StateAlias', 'target': 'Test State',
+ 'command': 'get state'}
+ test(): {'sender': 'Test State', '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.
+ """
+
+ async def receive(self, message: Message) -> None:
+ """Translate states from and commands to aliased state."""
+ alias_message = Message(self.name)
+ if ('target' in message and message['target'] == self.name and
+ 'command' in message):
+ alias_message['target'] = self.conf['alias for']
+ if message['command'] == 'get state':
+ alias_message['command'] = 'get state'
+ await self.bus.send(alias_message)
+ elif (message['command'] == 'set state' and
+ 'new state' in message):
+ alias_message['command'] = 'set state'
+ alias_message['new state'] = message['new state']
+ await self.bus.send(alias_message)
+ if (message['sender'] == self.conf['alias for'] and
+ 'state' in message):
+ if 'event' in message and message['event'] == 'changed':
+ alias_message['event'] = 'changed'
+ alias_message['state'] = message['state']
+ await self.bus.send(alias_message)
+
+ def process_conf(self) -> None:
+ """Register plugin as bus client."""
+ sends = [MessageTemplate({'target': {'const': self.conf['alias for']},
+ 'command': {'const': 'get state'}}),
+ MessageTemplate({'target': {'const': self.conf['alias for']},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}),
+ MessageTemplate({'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}}),
+ MessageTemplate({'state': {'type': 'boolean'}})]
+ receives = [MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'get state'}}),
+ MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}),
+ MessageTemplate({'sender':
+ {'const': self.conf['alias for']},
+ 'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}}),
+ MessageTemplate({'sender':
+ {'const': self.conf['alias for']},
+ 'state': {'type': 'boolean'}})]
+ self.bus.register(self.name, 'StateAlias',
+ sends, receives, self.receive)
+
+ async def run(self) -> None:
+ """Run no code proactively."""
+ pass
+
+
+class AndState(BasePlugin):
+ """Implement 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"}]))
+ ... # doctest: +NORMALIZE_WHITESPACE
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State 1', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State 1'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 1'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State 2', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State 2'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 2'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test AndState', 'plugin': 'AndState',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test AndState'},
+ 'command': {'const': 'get state'}},
+ {'sender': {'const': 'Test State 1'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State 2'},
+ 'state': {'type': 'boolean'}}]}
+ test(): {'sender': 'test()', 'target': 'Test State 1',
+ 'command': 'set state', 'new state': True}
+ 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 State 2', 'event': 'changed', 'state': True}
+ test(): {'sender': 'Test AndState', 'event': 'changed', 'state': True}
+ test(): {'sender': 'test()', 'target': 'Test State 1',
+ 'command': 'set state', 'new state': False}
+ test(): {'sender': 'Test State 1', 'event': 'changed', 'state': False}
+ test(): {'sender': 'Test AndState', 'event': 'changed', 'state': False}
+ test(): {'sender': 'test()', 'target': 'Test AndState',
+ 'command': 'get state'}
+ test(): {'sender': 'Test AndState', 'state': False}
+ """
+
+ 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.
+ """
+
+ async def receive(self, message: Message) -> None:
+ """."""
+ if ('target' in message and message['target'] == self.name and
+ 'command' in message and message['command'] == 'get state'):
+ await self.bus.send(Message(self.name, {'state': self.state}))
+ if 'state' in message and message['sender'] in self.conf['states']:
+ assert isinstance(message['sender'], str)
+ assert isinstance(message['state'], bool)
+ self.states[message['sender']] = message['state']
+ new_state = all(self.states.values())
+ if self.state != new_state:
+ self.state: bool = new_state
+ await self.bus.send(Message(self.name,
+ {'event': 'changed',
+ 'state': self.state}))
+
+ def process_conf(self) -> None:
+ """Register plugin as bus client."""
+ sends = [MessageTemplate({'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}}),
+ MessageTemplate({'state': {'type': 'boolean'}})]
+ receives = [MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'get state'}})]
+ self.states: Dict[str, bool] = {}
+ for state in self.conf['states']:
+ receives.append(MessageTemplate({'sender': {'const': state},
+ 'state': {'type': 'boolean'}}))
+ self.states[state] = False
+ self.state = all(self.states.values())
+ self.bus.register(self.name, 'AndState',
+ sends, receives, self.receive)
+
+ async def run(self) -> None:
+ """Run no code proactively."""
+ pass
+
+
+class OrState(BasePlugin):
+ """Implement 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"}]))
+ ... # doctest: +NORMALIZE_WHITESPACE
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State 1', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State 1'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 1'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test State 2', 'plugin': 'State',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test State 2'},
+ 'command': {'const': 'get state'}},
+ {'target': {'const': 'Test State 2'},
+ 'command': {'const': 'set state'},
+ 'new state': {'type': 'boolean'}}]}
+ test(): {'sender': '', 'event': 'registered',
+ 'client': 'Test OrState', 'plugin': 'OrState',
+ 'sends': [{'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}},
+ {'state': {'type': 'boolean'}}],
+ 'receives': [{'target': {'const': 'Test OrState'},
+ 'command': {'const': 'get state'}},
+ {'sender': {'const': 'Test State 1'},
+ 'state': {'type': 'boolean'}},
+ {'sender': {'const': 'Test State 2'},
+ 'state': {'type': 'boolean'}}]}
+ test(): {'sender': 'test()', 'target': 'Test State 1',
+ 'command': 'set state', 'new state': True}
+ test(): {'sender': 'Test State 1', 'event': 'changed', 'state': True}
+ test(): {'sender': 'Test OrState', 'event': 'changed', 'state': True}
+ test(): {'sender': 'test()', 'target': 'Test State 2',
+ 'command': 'set state', 'new state': True}
+ 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 State 1', 'event': 'changed', 'state': False}
+ test(): {'sender': 'test()', 'target': 'Test OrState',
+ 'command': 'get state'}
+ test(): {'sender': 'Test OrState', 'state': True}
+ """
+
+ 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.
+ """
+
+ async def receive(self, message: Message) -> None:
+ """."""
+ if ('target' in message and message['target'] == self.name and
+ 'command' in message and message['command'] == 'get state'):
+ await self.bus.send(Message(self.name, {'state': self.state}))
+ if 'state' in message and message['sender'] in self.conf['states']:
+ assert isinstance(message['sender'], str)
+ assert isinstance(message['state'], bool)
+ self.states[message['sender']] = message['state']
+ new_state = any(self.states.values())
+ if self.state != new_state:
+ self.state: bool = new_state
+ await self.bus.send(Message(self.name,
+ {'event': 'changed',
+ 'state': self.state}))
+
+ def process_conf(self) -> None:
+ """Register plugin as bus client."""
+ sends = [MessageTemplate({'event': {'const': 'changed'},
+ 'state': {'type': 'boolean'}}),
+ MessageTemplate({'state': {'type': 'boolean'}})]
+ receives = [MessageTemplate({'target': {'const': self.name},
+ 'command': {'const': 'get state'}})]
+ self.states: Dict[str, bool] = {}
+ for state in self.conf['states']:
+ receives.append(MessageTemplate({'sender': {'const': state},
+ 'state': {'type': 'boolean'}}))
+ self.states[state] = False
+ self.state = any(self.states.values())
+ self.bus.register(self.name, 'OrState',
+ sends, receives, self.receive)
async def run(self) -> None:
+ """Run no code proactively."""
pass
git.graph-it.com / graphit / controlpi.git / commitdiff