Module pytrap

TRAP extension for python3 (pytrap).

This module can be used to write NEMEA module. It consists of two main classes: TrapCtx and UnirecTemplate. TrapCtx contains communication interface, actually, it is a wrapper for libtrap. UnirecTemplate can be used for data access and manipulation. It uses UniRec macros and functions in order to retrieve value of a field and to store value into data message.

Simple example to receive and send one message:

import pytrap
c = pytrap.TrapCtx()
c.init(["-i", "u:socket1,u:socket2"], 1, 1)
fmttype = pytrap.FMT_UNIREC
fmtspec = "ipaddr SRC_IP"
c.setRequiredFmt(0, fmttype, fmtspec)
rec = pytrap.UnirecTemplate(fmtspec)
try:
    data = c.recv()
except pytrap.FormatChanged as e:
    fmttype, fmtspec = c.getDataFmt(0)
    rec = pytrap.UnirecTemplate(fmtspec)
    data = e.data
c.setDataFmt(0, fmttype, fmtspec)
rec.setData(data)
print(rec.strRecord())
# send the message that was received:
c.send(data)
c.finalize()

Simple example for data access using rec - UnirecTemplate instance:

print(rec.SRC_IP)
rec.SRC_IP = pytrap.UnirecIPAddr("127.0.0.1")
print(getattr(rec, "SRC_IP"))
rec.TIME_FIRST = pytrap.UnirecTime(12345678)
print(rec.TIME_FIRST)
print(rec.TIME_FIRST.toDatetime())

Simple example for creation of new message of UnirecTemplate:

# 100 is the maximal total size of fields with variable length
data = rec.createMessage(100)
rec.DST_PORT = 80

createMessage() should be called just at the beginning of program or when format change is needed.

There is a complete example module: https://github.com/CESNET/Nemea-Framework/tree/master/examples/python

For more details, see docstring of the classes and methods.

Functions

def getTrapVersion(...)

Get the version of libtrap.

Returns

tuple(str, str): libtrap version, git version.

Classes

class FormatChanged (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • TrapError
  • builtins.Exception
  • builtins.BaseException
class FormatMismatch (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • TrapError
  • builtins.Exception
  • builtins.BaseException
class Terminated (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • TrapError
  • builtins.Exception
  • builtins.BaseException
class TimeoutError (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • TrapError
  • builtins.Exception
  • builtins.BaseException
class TrapCtx

Class represents libtrap context. It must be initialized using init() and finalized and the end of usage finalize(). To terminate blocking libtrap functions use terminate().

Methods

def finalize(...)

Free allocated memory.

def getDataFmt(...)

Get data format that was negotiated via input IFC.

Args

ifcidx : Optional[int]
Index of IFC (default: 0).

Returns

Tuple(int, string): Type of format and specifier (see setRequiredFmt()).

def getInIFCState(...)

Get the state of input IFC.

Args

ifcidx : int
Index of IFC.

Returns

int
One of module's constants (FMTS_WAITING, FMTS_OK, FMTS_MISMATCH, FMTS_CHANGED).

Raises

TrapError
Bad index is passed or TRAP is not initialized.
def getVerboseLevel(...)

Get current verbose level.

Returns

int
Level of verbosity. See setVerboseLevel() for possible values.
def ifcctl(...)

Change settings of TRAP IFC.

Args

ifcidx : int
Index of IFC.
dir_in : bool
If True, input IFC will be modified, output IFC otherwise.
request : int
Type of request given by a module's constant (CTL_AUTOFLUSH, CTL_BUFFERSWITCH, CTL_TIMEOUT).
value : int
Parameter value of the chosen request.
def init(...)

Initialization of TRAP.

Args

argv : list[str]
Arguments of the process.
ifcin : Optional[int]
ifcin is a number of input IFC (default: 1).
ifcout : Optional[int]
ifcout is a number of output IFC (default: 0).
module_name : Optional[str]
Set name of this module.
module_desc : Optional[str]
Set description of this module.
service_ifcname : Optional[str]
Set identifier of service IFC, PID is used when omitted, set to "" (empty string) to disable service IFC.

Raises

TrapError
Initialization failed.
def recv(...)

Receive data via TRAP interface.

Args

ifcidx : Optional[int]
Index of input IFC (default: 0).

Returns

bytearray
Received data.

Raises

TimeoutError
Receiving data failed due to elapsed timeout.
TrapError
Bad index given.
FormatChanged
Data format was changed, it is necessary to update template. The received data is in data attribute of the FormatChanged instance.
Terminated
The TRAP IFC was terminated.
def send(...)

Send data via TRAP interface.

Args

bytes : bytearray or bytes
Data to send.
ifcidx : Optional[int]
Index of output IFC (default: 0).

Raises

TimeoutError
Receiving data failed due to elapsed timeout.
TrapError
Bad size or bad index given.
Terminated
The TRAP IFC was terminated.
def sendFlush(...)

Force sending buffer for IFC with ifcidx index.

Args

ifcidx : Optional[int]
Index of IFC (default: 0).
def setDataFmt(...)

Set data format for output IFC.

Args

ifcidx : int
Index of IFC.
type : Optional[int]
Type of format (FMT_RAW, FMT_UNIREC), (default: FMT_UNIREC).
spec : Optional[string]
Specifier of data format (default: "").
def setRequiredFmt(...)

Set required data format for input IFC.

Args

ifcidx : int
Index of IFC.
type : Optional[int]
Type of format: FMT_RAW, FMT_UNIREC (default: FMT_UNIREC)
spec : Optional[string]
Specifier of data format (UniRec specifier for FMT_UNIREC type) (default: "").
def setVerboseLevel(...)

Set the verbose level of TRAP.

Args

level : int
Level of verbosity the higher value the more verbose. Possible values: VERB_ERRORS (-3), VERB_WARNINGS (-2), VERB_NOTICES (-1) = default, VERB_VERBOSE (0), VERB_VERBOSE2 (1), VERB_VERBOSE3 (2).
def terminate(...)

Terminate TRAP.

class TrapError (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • builtins.Exception
  • builtins.BaseException

Subclasses

class TrapHelp (*args, **kwargs)

Common base class for all non-exit exceptions.

Ancestors

  • builtins.Exception
  • builtins.BaseException
class UnirecIPAddr (*args, **kwargs)

UnirecIPAddr(ip) Class for UniRec IP Address storage and base data access.

Args

ip : str
text represented IPv4 or IPv6 address

Methods

def dec(...)

Decrement IP address.

Returns

UnirecIPAddr
New decremented IPAddress.
def inc(...)

Increment IP address.

Returns

UnirecIPAddr
New incremented IPAddress.
def isIPv4(...)

Check if the address is IPv4.

Returns

bool
True if the address is IPv4.
def isIPv6(...)

Check if the address is IPv6.

Returns

bool
True if the address is IPv6.
def isNull(...)

Check if the address is null (IPv4 or IPv6), i.e. "0.0.0.0" or "::".

Returns

bool
True if the address is null.
class UnirecIPAddrRange (*args, **kwargs)

UnirecIPAddrRange(start, [end]) Class for UniRec IP Address Range storage and base data access.

Args

start : str
text represented IPv4 address or prefix using /mask notation
end : Optional[str]
text represented IPv4 address of end of the range

(start must not contain /mask)

Instance variables

var end

High IP address of range

var start

Low IP address of range

Methods

def isIn(...)

Check if the address is in IP range.

Args

ipaddr
UnirecIPAddr structReturns:
long
-1 if ipaddr < start of range, 0 if ipaddr is in interval, 1 if ipaddr > end of range .
def isOverlap(...)

Check if 2 ranges sorted by start IP overlap.

Args

other
UnirecIPAddrRange, second interval to compareReturns:
bool
True if ranges are overlaps.
class UnirecMACAddr (*args, **kwargs)

UnirecMACAddr(mac) Class for UniRec MAC Address storage and base data access.

Args

mac : str
text represented MAC address (e.g. "00:11:22:33:44:55")

Methods

def dec(...)

Decrement MAC address.

Returns

UnirecMACAddr
New decremented MAC address.
def inc(...)

Increment MAC address.

Returns

UnirecMACAddr
New incremented MAC address.
def isNull(...)

Check if the MAC address is null, i.e. "00:00:00:00:00:00".

Returns

bool
True if the address is null.
class UnirecMACAddrRange (*args, **kwargs)

UnirecMACAddrRange(start, end) Class for UniRec MAC Address Range storage and base data access.

Args

start : UnirecMACAddr
text represented MAC address or prefix using /mask notation
end : UnirecMACAddr
text represented MAC address of end of the range

(start must not contain /mask)

Instance variables

var end

High MAC address of range

var start

Low MAC address of range

Methods

def isIn(...)

Check if the address is in MAC range.

Args

macaddr
UnirecMACAddr structReturns:
long
-1 if macaddr < start of range, 0 if macaddr is in interval, 1 if macaddr > end of range .
def isOverlap(...)

Check if 2 ranges sorted by start MAC overlap.

Args

other
UnirecMACAddrRange, second interval to compareReturns:
bool
True if ranges are overlaps.
class UnirecTemplate (...)

UnirecTime(spec) Class for UniRec template storage and base data access.

Example

UnirecTemplate("ipaddr SRC_IP,uint16 SRC_PORT,time START") creates a template with tree fields.

Args

spec : str
UniRec template specifier - list of field types and names

Methods

def copy(...)

Create a new instance with the same format specifier without data.

Returns

UnirecTemplate
New copy of object (not just reference).
def createMessage(...)

Create a message that can be filled in with values according to the template.

Args

dyn_size : Optional[int]
Maximal size of variable data (in total) (default: 0).

Returns

bytearray
Allocated memory that can be filled in using set().
def get(...)

Get value of the field from the UniRec message.

Args

data : bytearray or bytes
Data - UniRec message.
field_name : str
Field name.

Returns

(object): Retrieved value of the field (depends on UniRec template).

Raises

TypeError
Data argument must be bytearray or bytes.
TrapError
Field name was not found.
def getByID(...)

Get value of the field from the UniRec message.

Args

data : bytearray or bytes
Data - UniRec message.
field_id : int
Field ID (use getFieldsDict()).

Returns

(object): Retrieved value of the field (depends on UniRec template).

Raises

TypeError
Data argument must be bytearray or bytes.
def getData(...)

Get data that was already set using setData.

Returns

bytearray
Data - UniRec message.
def getFieldType(...)

Get type of given field.

Args

field_name : str
Field name.

Returns

type
Type object (e.g. int, str or pytrap.UnirecIPAddr).
def getFieldsDict(...)

Get set of fields of the template.

Returns

Dict(str,int): Dictionary of field_id with field name as a key.

def recFixlenSize(...)

Get size of fixed part of UniRec record.

This is the minimal size of a record with given template,i.e. the size with all variable-length fields empty.

Returns

int
Size of fixed part of template in bytes
def recSize(...)

Get size of UniRec record.

Return total size of valid record data, i.e. number of bytes occupied by all fields.This may be less than allocated size of 'data'.

Args

data : Optional[bytearray or bytes]
Data of UniRec message (optional if previously set by setData)Returns:
int
Size of record data in bytes
def recVarlenSize(...)

Get size of variable-length part of UniRec record.

Return total size of all variable-length fields.

Args

data : Optional[bytearray or bytes]
Data of UniRec message (optional if previously set by setData)Returns:
int
Total size of variable-length fields in bytes
def set(...)

Set value of the field in the UniRec message.

Args

data : bytearray or bytes
Data - UniRec message.
field_name : str
Field name.
value : object
New value of the field (depends on UniRec template).

Raises

TypeError
Bad object type of value was given.
TrapError
Field was not found.
def setByID(...)

Set value of the field in the UniRec message.

Args

data : bytearray or bytes
Data - UniRec message.
field_id : int
Field ID.
value : object
New value of the field (depends on UniRec template).

Raises

TypeError
Bad object type of value was given.
TrapError
Field was not found.
def setData(...)

Set data for attribute access.

Args

data : bytearray or bytes
Data - UniRec message.
def strRecord(...)

Get values of record in readable format.

Returns

str
String in key=value format.
class UnirecTime (*args, **kwargs)

UnirecTime(int(seconds), [int(miliseconds)]) UnirecTime(double(secs_and_msecs)) UnirecTime(str("2019-03-18T12:11:10Z")) Class for UniRec timestamp storage and base data access.

Args:
    str: datetime, e.g., "2019-03-18T12:11:10.123Z"
    double or int: number of seconds
    Optional[int]: number of miliseconds (when the first argument is int)

Raises:
    TypeError: unsupported type was provided or string is malformed.

Static methods

def fromDatetime(...)

Get UnirecTime from a datetime object.

Returns

(UnirecTime): Retrieved timestamp.

def now(...)

Get UnirecTime instance of current time.

Returns

(UnirecTime): Current date and time.

Methods

def format(...)

Get timestamp as a datetime object.

Args

format : Optional[str]
Formatting string, same as for datetime.strftime (default: "%FT%TZ").

Returns

(str): Formatted timestamp as string.

def getMiliSeconds(...)

Get number of seconds of timestamp.

Returns

(long): Retrieved number of seconds.

def getSeconds(...)

Get number of seconds of timestamp.

Returns

(long): Retrieved number of seconds.

def getTimeAsFloat(...)

Get number of seconds of timestamp.

Returns

(double): Retrieved timestamp as floating point number.

def toDatetime(...)

Get timestamp as a datetime object.

Returns

(datetime): Retrieved timestamp as datetime.