Module 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.
All these plugins use the following conventions:
- If their state changes they send a message containing "event": "changed"
and "state":
. - If their state is reported due to a message, but did not change they send
a message containing just "state":
. - If they receive a message containing "target":
and "command": "get state" they report their current state. - If State (or any other settable state using these conventions) receives
a message containing "target":
, "command": "set state" and "new state": 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 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}
Expand source code
"""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.
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.
- 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 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, 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':
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']
await self.bus.send(Message(self.name,
{'event': 'changed',
'state': self.state}))
else:
await self.bus.send(Message(self.name,
{'state': self.state}))
def process_conf(self) -> None:
"""Register plugin as bus client."""
self.state = False
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:
"""Process "get state" command and messages of combined states."""
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:
"""Process "get state" command and messages of combined states."""
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."""
passClasses
class State (bus: MessageBus, name: str, conf: Dict[str, Any])-
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}Expand source code
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': 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'] await self.bus.send(Message(self.name, {'event': 'changed', 'state': self.state})) else: await self.bus.send(Message(self.name, {'state': self.state})) def process_conf(self) -> None: """Register plugin as bus client.""" self.state = False 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.""" passAncestors
- BasePlugin
- abc.ABC
Class variables
var CONF_SCHEMA-
Schema for State plugin configuration.
There are no required or optional configuration keys.
Methods
async def receive(self, message: Message) ‑> NoneType-
Process commands to get/set state.
Expand source code
async def receive(self, message: Message) -> None: """Process commands to get/set state.""" if message['command'] == 'get state': 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'] await self.bus.send(Message(self.name, {'event': 'changed', 'state': self.state})) else: await self.bus.send(Message(self.name, {'state': self.state})) def process_conf(self) ‑> NoneType-
Register plugin as bus client.
Expand source code
def process_conf(self) -> None: """Register plugin as bus client.""" self.state = False 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) ‑> NoneType-
Run no code proactively.
Expand source code
async def run(self) -> None: """Run no code proactively.""" pass
class StateAlias (bus: MessageBus, name: str, conf: Dict[str, Any])-
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}Expand source code
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.""" passAncestors
- BasePlugin
- abc.ABC
Class variables
var CONF_SCHEMA-
Schema for StateAlias plugin configuration.
Required configuration key:
- 'alias for': name of aliased state.
Methods
async def receive(self, message: Message) ‑> NoneType-
Translate states from and commands to aliased state.
Expand source code
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) ‑> NoneType-
Register plugin as bus client.
Expand source code
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) ‑> NoneType-
Run no code proactively.
Expand source code
async def run(self) -> None: """Run no code proactively.""" pass
class AndState (bus: MessageBus, name: str, conf: Dict[str, Any])-
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}Expand source code
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: """Process "get state" command and messages of combined states.""" 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.""" passAncestors
- BasePlugin
- abc.ABC
Class variables
var CONF_SCHEMA-
Schema for AndState plugin configuration.
Required configuration key:
- 'states': list of names of combined states.
Methods
async def receive(self, message: Message) ‑> NoneType-
Process "get state" command and messages of combined states.
Expand source code
async def receive(self, message: Message) -> None: """Process "get state" command and messages of combined states.""" 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) ‑> NoneType-
Register plugin as bus client.
Expand source code
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) ‑> NoneType-
Run no code proactively.
Expand source code
async def run(self) -> None: """Run no code proactively.""" pass
class OrState (bus: MessageBus, name: str, conf: Dict[str, Any])-
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}Expand source code
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: """Process "get state" command and messages of combined states.""" 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.""" passAncestors
- BasePlugin
- abc.ABC
Class variables
var CONF_SCHEMA-
Schema for OrState plugin configuration.
Required configuration key:
- 'states': list of names of combined states.
Methods
async def receive(self, message: Message) ‑> NoneType-
Process "get state" command and messages of combined states.
Expand source code
async def receive(self, message: Message) -> None: """Process "get state" command and messages of combined states.""" 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) ‑> NoneType-
Register plugin as bus client.
Expand source code
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) ‑> NoneType-
Run no code proactively.
Expand source code
async def run(self) -> None: """Run no code proactively.""" pass