Source code for pycoin.message.PeerAddress

import functools
import struct

from pycoin.intbytes import iterbytes
from pycoin.satoshi.satoshi_struct import parse_struct
from pycoin.encoding.hexbytes import h2b


IP4_HEADER = h2b("00000000000000000000FFFF")


[docs]def ip_bin_to_ip6_addr(ip_bin): return ":".join("%x" % v for v in struct.unpack(">HHHHHHHH", ip_bin))
[docs]def ip_bin_to_ip4_addr(ip_bin): return "%d.%d.%d.%d" % tuple(iterbytes(ip_bin[-4:]))
[docs]@functools.total_ordering class PeerAddress(object): def __init__(self, services, ip_bin, port): self.services = int(services) assert isinstance(ip_bin, bytes) if len(ip_bin) == 4: ip_bin = IP4_HEADER + ip_bin assert len(ip_bin) == 16 self.ip_bin = ip_bin self.port = port def __repr__(self): return "%s/%d" % (self.host(), self.port)
[docs] def host(self): if self.ip_bin.startswith(IP4_HEADER): return ip_bin_to_ip4_addr(self.ip_bin[-4:]) return ip_bin_to_ip6_addr(self.ip_bin)
[docs] def stream(self, f): f.write(struct.pack("<Q", self.services)) f.write(self.ip_bin) f.write(struct.pack("!H", self.port))
[docs] @classmethod def parse(self, f): services, ip_bin, port = parse_struct("Q@h", f) self.ip_bin = ip_bin return self(services, self.ip_bin, port)
def __lt__(self, other): return (self.ip_bin, self.port, self.services) < (other.ip_bin, other.port, other.services) def __eq__(self, other): return self.services == other.services and \ self.ip_bin == other.ip_bin and self.port == other.port