pyebus.msgdef module

Message Defintions.

class pyebus.msgdef.MsgDef(circuit, name, children, read=False, prio=None, write=False, update=False, setprio=None)[source]

Bases: _MsgDef, NodeMixin

Message Definition.

A message is defined by circuit and name and has a fields

Parameters:
Keyword Arguments:

  • read (bool) – Message intend to be read

  • prio (int) – Message Polling Priority on bus.

  • write (bool) – Message intend to be written

  • updated (bool) – Message intent to be seen automatically on every value change

  • setprio – Message polling priority to be set. Integer 1-9 or `A for automatic.

>>> from pyebus import MsgDef, types
>>> m = MsgDef('circuit', 'name', [
...     FieldDef(0, 'temp', types.IntType(-127, 128)),
...     FieldDef(1, 'name', types.StrType(10)),
...     VirtFieldDef('virt', types.StrType(), lambda msg: f'{msg.name}: {msg.temp}'),
... ], read=True)
>>> m  
MsgDef('circuit', 'name', (FieldDef(0, 'temp', IntType(... VirtFieldDef('virt', StrType())), read=True)
>>> m.ident
'circuit/name'
>>> m.children
(FieldDef(0, 'temp', IntType(-127, 128)), FieldDef(1, 'name', StrType(length=10)), VirtFieldDef('virt', StrType()))
>>> m.fields
(FieldDef(0, 'temp', IntType(-127, 128)), FieldDef(1, 'name', StrType(length=10)))
>>> m.virtfields
(VirtFieldDef('virt', StrType()),)
>>> m.access
'r----'

MsgDef, FieldDef and VirtFieldDef form a tree structure

>>> from anytree import RenderTree
>>> print(RenderTree(m))  
MsgDef('circuit', 'name', (FieldDef(0, 'temp', IntType(-127, 128)),..., StrType())), read=True)
├── FieldDef(0, 'temp', IntType(-127, 128))
├── FieldDef(1, 'name', StrType(length=10))
└── VirtFieldDef('virt', StrType())

Similar object can be easily created by:

>>> m.replace(children=m.fields[1:2])
MsgDef('circuit', 'name', (FieldDef(1, 'name', StrType(length=10)),), read=True)

Create new instance of _MsgDef(circuit, name, read, prio, write, update, setprio)

property fields

Fields.

property virtfields

Calulated Fields.

property ident

Identifier.

property access

Message Access.

join(msgdef)[source]

Return Joined Message Definition.

replace(**kwargs)[source]

Create copy with updated attributes.

class pyebus.msgdef.AbstractFieldDef(idx, name, type_, unit=None, comment=None)[source]

Bases: _FieldDef, NodeMixin

Abstract Field Definition.

Parameters:

  • idx (str) – Index within Message

  • name (str) – Unique name (as name may be used multiple times by ebus)

  • type (Type) – Type

Keyword Arguments:

  • unit (str) – Unit of the field value

  • comment (str) – Comment.

Create new instance of _FieldDef(idx, name, type_, unit, comment)

property msgdef

Message Definition.

property ident

Identifier.

class pyebus.msgdef.FieldDef(idx, name, type_, unit=None, comment=None)[source]

Bases: AbstractFieldDef

Field Definition.

Parameters:

  • idx (str) – Index within Message

  • name (str) – Unique name (as name may be used multiple times by ebus)

  • type (Type) – Type

Keywords Args:

unit (str): Unit of the field value comment (str): Comment.

Create new instance of _FieldDef(idx, name, type_, unit, comment)

class pyebus.msgdef.VirtFieldDef(name, type_, func, unit=None, comment=None)[source]

Bases: AbstractFieldDef

Virtual Field Definition.

Parameters:

  • name (str) – Unique name (as name may be used multiple times by ebus)

  • type (Type) – Type

  • func – Function to create value.

Keywords Args:

unit (str): Unit of the field value comment (str): Comment.

Create new instance of _FieldDef(idx, name, type_, unit, comment)

pyebus.msgdef.resolve_prio(msgdef, setprio='A')[source]

Resolve priority specification.

Integer values are just passed through.

>>> resolve_prio(MsgDef('circuit', 'name', [], read=True), 1)
1
>>> resolve_prio(MsgDef('circuit', 'name', [], read=True), 9)
9

The AUTO option, sets read-only messages to prio 1.

>>> resolve_prio(MsgDef('circuit', 'name', [], read=True), AUTO)
1

All other readable messages get prio 2

>>> resolve_prio(MsgDef('circuit', 'name', [], read=True, write=True), AUTO)
2
>>> resolve_prio(MsgDef('circuit', 'name', [], read=True, update=True), AUTO)
2

Non-readable messages have no prio.

>>> resolve_prio(MsgDef('circuit', 'name', [], write=True), AUTO)