duct.protocol

duct.protocol.elasticsearch

class duct.protocol.elasticsearch.ElasticSearch(url='http://localhost:9200', user=None, password=None, index='duct-%Y.%m.%d')[source]

Bases: object

Elasticsearch HTTP API client

async bulkIndex(rows)[source]

Insert many indicies

async insertIndex(index_type, data)[source]

Insert an index

async node_stats()[source]

Return statistics for this node

async stats()[source]

Return statistics for the cluster

duct.protocol.i2c

class duct.protocol.i2c.I2CTransaction(bus: int, device: int)[source]

Bases: object

MPL115 temperature and pressure sensor

Arguments: :param smbus: Bus number (default 1) :type smbus: int. :param device: I2C device address :type device: int.

close()[source]

Close the transaction

async read(n_bytes: int) → bytes[source]

Read n_bytes

async read_block(register: int, length: int) → list[int][source]

Read a register block

async read_byte() → int[source]

Read one byte

async read_byte_data(register: int) → int[source]

Read a register byte

async transfer(write_data: bytes, read_n: int) → bytes[source]

Write then immediately read in a single combined transaction.

async write(data: bytes)[source]

Write raw data

async write_block(register: int, data: list[int])[source]

Write a register block

async write_byte(value: int)[source]

Write one byte

async write_byte_data(register: int, value: int)[source]

Write a register byte

duct.protocol.i2c.list_buses() → list[int][source]

Lists available i2c busses

duct.protocol.i2c.transaction(bus: int, device: int)[source]

Acquire exclusive access to a bus and yield an I2CTransaction bound to device.

Callers on the same bus number are queued and served in order; callers on different buses run concurrently.

duct.protocol.icmp

class duct.protocol.icmp.EchoPacket(seq=0, eid=None, data=None, packet=None)[source]

Bases: object

ICMP Echo packet encoder and decoder

calculateChecksum(buf)[source]

Calculate ICMP checksum over buf

decodePacket(packet)[source]

Decode an ICMP echo reply packet

encodePacket()[source]

Encode an ICMP echo request packet

class duct.protocol.icmp.IP(packet)[source]

Bases: object

IP header decoder

readPacket(packet)[source]

Decode an IP packet header

async duct.protocol.icmp.ping(dst, count, inter=0.2, maxwait=1000, size=64)[source]

Send ICMP echo requests to dst count times.

Returns (packet_loss_percent, avg_latency_ms) or (100, None) on failure.

Requires CAP_NET_RAW or root privileges.

duct.protocol.influxdb3

class duct.protocol.influxdb3.InfluxDB3Client(url='http://localhost:8086', database='duct', token=None, precision='nanosecond')[source]

Bases: object

InfluxDB 3 HTTP write client

async write(lines)[source]

Write line protocol data to InfluxDB 3.

Parameters:

lines – List of line protocol strings

duct.protocol.influxdb3.to_line_protocol(measurement, tags, fields, timestamp_ns)[source]

Encode a single point in InfluxDB line protocol.

Parameters:

  • measurement – Measurement name

  • tags – Dict of tag key/value pairs

  • fields – Dict of field key/value pairs

  • timestamp_ns – Timestamp in nanoseconds

Returns:

Line protocol string (no trailing newline)

duct.protocol.opentsdb

class duct.protocol.opentsdb.OpenTSDBClient(url='http://localhost:4242', user=None, password=None)[source]

Bases: object

OpenTSDB HTTP client

put(data)[source]

Put one or more metrics

duct.protocol.riemann

class duct.protocol.riemann.RiemannProtobufMixin[source]

Bases: object

Class mix-in for protocol buffer encoding/decoding

decodeMessage(data)[source]

Decode a protobuf message

encodeEvent(event)[source]

Adapt an Event object to a Riemann protobuf Event

encodeMessage(events)[source]

Encode a list of Duct events as a protobuf message

class duct.protocol.riemann.RiemannTCPClient(host, port, failover=False)[source]

Bases: RiemannProtobufMixin

Asyncio TCP Riemann client with automatic reconnection.

Parameters:

  • host – Riemann server hostname or list of hostnames for failover

  • port – Riemann server port

  • failover – Enable round-robin failover across host list

async connect()[source]

Establish TCP connection to the current host

async disconnect()[source]

Close the connection

async reconnect_loop(delay=5.0, max_delay=30.0)[source]

Keep trying to reconnect with exponential backoff

async sendEvents(events)[source]

Send a list of Duct events to Riemann. Fire-and-forget on error.

class duct.protocol.riemann.RiemannUDPProtocol(host, port, on_ready=None)[source]

Bases: RiemannProtobufMixin, DatagramProtocol

Asyncio UDP datagram protocol for Riemann

connection_lost(exc)[source]

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

connection_made(transport)[source]

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

error_received(exc)[source]

Called when a send or receive operation raises an OSError.

(Other than BlockingIOError or InterruptedError.)

sendEvents(events)[source]

Send events as a UDP datagram (best-effort)

async duct.protocol.riemann.create_riemann_udp(host, port)[source]

Create and return a RiemannUDPProtocol bound to an ephemeral port

duct.protocol.senml

duct.protocol.senml.event_to_json(event: Event) → bytes[source]

Serialise a Duct Event to plain JSON bytes using the native event dict.

duct.protocol.senml.event_to_senml(event: Event) → bytes[source]

Serialise a Duct Event to SenML-JSON bytes.

duct.protocol.senml.event_to_senml_cbor(event: Event) → bytes[source]

Serialise a Duct Event to SenML-CBOR bytes.

duct.protocol.senml.event_to_senml_pack(event: Event) → list[source]

Convert a Duct Event to a SenML record pack (list of dicts).

duct.protocol.senml.json_to_event(data: bytes | str, default_ttl: float = 60.0) → Event[source]

Deserialise a plain JSON event dict to a Duct Event.

duct.protocol.senml.senml_cbor_to_event(data: bytes, default_ttl: float = 60.0) → Event[source]

Deserialise SenML-CBOR bytes to a Duct Event.

duct.protocol.senml.senml_pack_to_event(pack: list, default_ttl: float = 60.0) → Event[source]

Convert a SenML record pack (list of dicts) to a Duct Event.

duct.protocol.senml.senml_to_event(data: bytes | str, default_ttl: float = 60.0) → Event[source]

Deserialise SenML-JSON bytes or string to a Duct Event.

duct.protocol.sflow.protocol.counters

class duct.protocol.sflow.protocol.counters.DiskIOCounters(u)[source]

Bases: object

Counters for disk IO

class duct.protocol.sflow.protocol.counters.EthernetCounters(u)[source]

Bases: object

Counters for ethernet frames

class duct.protocol.sflow.protocol.counters.HostAdapters(u)[source]

Bases: object

Counters for HBAs

class duct.protocol.sflow.protocol.counters.HostCPUCounters(u)[source]

Bases: object

Counters for host CPU stats

class duct.protocol.sflow.protocol.counters.HostCounters(u)[source]

Bases: object

Counters for host endpoints

class duct.protocol.sflow.protocol.counters.HostMemoryCounters(u)[source]

Bases: object

Counters for host memory

class duct.protocol.sflow.protocol.counters.HostParent(u)[source]

Bases: object

Counters for hosts

class duct.protocol.sflow.protocol.counters.InterfaceCounters(u)[source]

Bases: object

Counters for network interfaces

class duct.protocol.sflow.protocol.counters.NetIOCounters(u)[source]

Bases: object

Counters for network interface IO

class duct.protocol.sflow.protocol.counters.SocketIPv4Counters(u)[source]

Bases: object

Counters for IPv4 sockets

class duct.protocol.sflow.protocol.counters.SocketIPv6Counters(u)[source]

Bases: object

Counters for IPv6 sockets

class duct.protocol.sflow.protocol.counters.TokenringCounters(u)[source]

Bases: object

Counters for Token ring networks

class duct.protocol.sflow.protocol.counters.VGCounters(u)[source]

Bases: object

Counters for AnyLan frames

class duct.protocol.sflow.protocol.counters.VLANCounters(u)[source]

Bases: object

Counters for VLANs

class duct.protocol.sflow.protocol.counters.VirtDiskIOCounters(u)[source]

Bases: object

Counters for virtual disk IO

class duct.protocol.sflow.protocol.counters.VirtMemoryCounters(u)[source]

Bases: object

Counters for virtual memory

class duct.protocol.sflow.protocol.counters.VirtNetIOCounters(u)[source]

Bases: object

Counters for virtual network adapters

duct.protocol.sflow.protocol.counters.getDecoder(fmt)[source]

Retrieve the decoder associated with a frame format ID

duct.protocol.sflow.protocol.flows

class duct.protocol.sflow.protocol.flows.EthernetSample(u)[source]

Bases: object

Ethernet sample data

class duct.protocol.sflow.protocol.flows.GatewaySample(u)[source]

Bases: object

IP gateway sample

class duct.protocol.sflow.protocol.flows.HeaderSample(u)[source]

Bases: object

Header sample data

class duct.protocol.sflow.protocol.flows.IEEE80211MACHeader(u)[source]

Bases: object

Ethernet MAC header decoder Not implemented yet

class duct.protocol.sflow.protocol.flows.IPV4Sample(u)[source]

Bases: object

IPv4 sample data

class duct.protocol.sflow.protocol.flows.IPV6Sample(u)[source]

Bases: object

IPv6 sample data

class duct.protocol.sflow.protocol.flows.IPv4Header(data)[source]

Bases: object

IPv4 header object

class duct.protocol.sflow.protocol.flows.IPv6Header(u)[source]

Bases: object

IPv6 header decoder object

class duct.protocol.sflow.protocol.flows.ISO8023Header(data)[source]

Bases: object

ISO8023 header object

class duct.protocol.sflow.protocol.flows.MPLSFTNSample(u)[source]

Bases: object

MPLS FTNS sample

class duct.protocol.sflow.protocol.flows.MPLSLDPFECSample(u)[source]

Bases: object

MPLS LDP FECS sample

class duct.protocol.sflow.protocol.flows.MPLSSample(u)[source]

Bases: object

MPLS sample

class duct.protocol.sflow.protocol.flows.MPLSTunnelSample(u)[source]

Bases: object

MPLS Tunnel sample

class duct.protocol.sflow.protocol.flows.MPLSVCSample(u)[source]

Bases: object

MPLS PVC sample

class duct.protocol.sflow.protocol.flows.NATSample(u)[source]

Bases: object

NAT sample

class duct.protocol.sflow.protocol.flows.PPPHeader(u)[source]

Bases: object

PPP header decoder Not implemented yet

class duct.protocol.sflow.protocol.flows.RouterSample(u)[source]

Bases: object

IP router sample

class duct.protocol.sflow.protocol.flows.SwitchSample(u)[source]

Bases: object

Ethernet switch sample data

class duct.protocol.sflow.protocol.flows.URLSample(u)[source]

Bases: object

URL sample

class duct.protocol.sflow.protocol.flows.UserSample(u)[source]

Bases: object

User sample

class duct.protocol.sflow.protocol.flows.VLANTunnelSample(u)[source]

Bases: object

VLAN tunnel sample

duct.protocol.sflow.protocol.flows.getDecoder(fmt)[source]

Return the decoder class for the given format

duct.protocol.sflow.protocol.protocol

class duct.protocol.sflow.protocol.protocol.CounterSample(u)[source]

Bases: object

Counter sample object

class duct.protocol.sflow.protocol.protocol.FlowSample(u)[source]

Bases: object

Flow sample object

class duct.protocol.sflow.protocol.protocol.Sflow(payload, host)[source]

Bases: object

SFlow protocol stream decoder

decode_samples(u)[source]

Decode samples received

sflow_v5(u)[source]

SFlow version 5 decoder

duct.protocol.sflow.protocol.utils

class duct.protocol.sflow.protocol.utils.IPv4Address(addr_int)[source]

Bases: object

IPv4 address object

asString()[source]

Return a human readable string for the IP

duct.protocol.sflow.protocol.utils.unpack_address(u)[source]

Unpack a binary network address

duct.protocol.sflow.protocol.xdr

Minimal XDR Unpacker replacing the deprecated xdrlib module.

class duct.protocol.sflow.protocol.xdr.Unpacker(data)[source]

Bases: object

Minimal XDR unpacker replacing the deprecated xdrlib module.

get_buffer()[source]

Return remaining unpacked bytes.

unpack_array(item_fn)[source]

Unpack an array using item_fn to decode each element.

unpack_float()[source]

Unpack a 32-bit float.

unpack_fopaque(n)

Unpack a fixed-length opaque byte string (padded to 4 bytes).

unpack_fstring(n)[source]

Unpack a fixed-length opaque byte string (padded to 4 bytes).

unpack_opaque()[source]

Unpack a variable-length opaque byte string.

unpack_string()[source]

Unpack a variable-length string.

unpack_uhyper()[source]

Unpack a 64-bit unsigned integer.

unpack_uint()[source]

Unpack a 32-bit unsigned integer.

unpack_unsigend()

Unpack a 32-bit unsigned integer.

duct.protocol.sflow.server

class duct.protocol.sflow.server.DatagramReceiver[source]

Bases: DatagramProtocol

asyncio DatagramProtocol for receiving sFlow packets

connection_made(transport)[source]

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

datagram_received(data, addr)[source]

Called when some datagram is received.

error_received(exc)[source]

Called when a send or receive operation raises an OSError.

(Other than BlockingIOError or InterruptedError.)

process_counter_sample(sflow, counter)[source]

Process an incoming counter sample

process_flow_sample(sflow, flow)[source]

Process an incoming flow sample

receive_counter(counter, host)[source]

Called when a counter is received

receive_flow(flow, sample, host)[source]

Called when a flow is received

receive_host_counter(counter, host)[source]

Called when a host counter is received

duct.protocol.ssh

class duct.protocol.ssh.SSHClient(hostname, username, port, password=None, knownhosts=None)[source]

Bases: object

Create an SSH connection and tunnel commands over it.

Parameters:

  • hostname – Hostname to connect to

  • username – Username to authenticate with

  • port – Port to connect to (default: 22)

  • password – Optional password for authentication

  • knownhosts – Known hosts file path

addKeyFile(kfile, password=None)[source]

Import a private key from a file

addKeyString(kstring, password=None)[source]

Import a private key from a string

async connect()[source]

Open connection to SSH server

async fork(command, args=(), env=None, path=None, _timeout=3600)[source]

Execute a remote command. Returns (stdout, stderr, exit_code).