Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/dns»

Browse Source
This commit is contained in:
Polina
2026-07-02 18:07:39 +00:00
parent 2b62519d62
commit 423067c703
5 changed files with 2934 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

355
venv/Lib/site-packages/dns/renderer.py Normal file
View File

@@ -0,0 +1,355 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2001-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""Help for building DNS wire format messages"""
import contextlib
import io
import random
import struct
import time
import dns.edns
import dns.exception
import dns.rdataclass
import dns.rdatatype
import dns.tsig
# Note we can't import dns.message for cicularity reasons
QUESTION = 0
ANSWER = 1
AUTHORITY = 2
ADDITIONAL = 3
@contextlib.contextmanager
def prefixed_length(output, length_length):
output.write(b"\00" * length_length)
start = output.tell()
yield
end = output.tell()
length = end - start
if length > 0:
try:
output.seek(start - length_length)
try:
output.write(length.to_bytes(length_length, "big"))
except OverflowError:
raise dns.exception.FormError
finally:
output.seek(end)
class Renderer:
"""Helper class for building DNS wire-format messages.
Most applications can use the higher-level L{dns.message.Message}
class and its to_wire() method to generate wire-format messages.
This class is for those applications which need finer control
over the generation of messages.
Typical use::
r = dns.renderer.Renderer(id=1, flags=0x80, max_size=512)
r.add_question(qname, qtype, qclass)
r.add_rrset(dns.renderer.ANSWER, rrset_1)
r.add_rrset(dns.renderer.ANSWER, rrset_2)
r.add_rrset(dns.renderer.AUTHORITY, ns_rrset)
r.add_rrset(dns.renderer.ADDITIONAL, ad_rrset_1)
r.add_rrset(dns.renderer.ADDITIONAL, ad_rrset_2)
r.add_edns(0, 0, 4096)
r.write_header()
r.add_tsig(keyname, secret, 300, 1, 0, '', request_mac)
wire = r.get_wire()
If padding is going to be used, then the OPT record MUST be
written after everything else in the additional section except for
the TSIG (if any).
output, an io.BytesIO, where rendering is written
id: the message id
flags: the message flags
max_size: the maximum size of the message
origin: the origin to use when rendering relative names
compress: the compression table
section: an int, the section currently being rendered
counts: list of the number of RRs in each section
mac: the MAC of the rendered message (if TSIG was used)
"""
def __init__(self, id=None, flags=0, max_size=65535, origin=None):
"""Initialize a new renderer."""
self.output = io.BytesIO()
if id is None:
self.id = random.randint(0, 65535)
else:
self.id = id
self.flags = flags
self.max_size = max_size
self.origin = origin
self.compress = {}
self.section = QUESTION
self.counts = [0, 0, 0, 0]
self.output.write(b"\x00" * 12)
self.mac = ""
self.reserved = 0
self.was_padded = False
def _rollback(self, where):
"""Truncate the output buffer at offset *where*, and remove any
compression table entries that pointed beyond the truncation
point.
"""
self.output.seek(where)
self.output.truncate()
keys_to_delete = []
for k, v in self.compress.items():
if v >= where:
keys_to_delete.append(k)
for k in keys_to_delete:
del self.compress[k]
def _set_section(self, section):
"""Set the renderer's current section.
Sections must be rendered order: QUESTION, ANSWER, AUTHORITY,
ADDITIONAL. Sections may be empty.
Raises dns.exception.FormError if an attempt was made to set
a section value less than the current section.
"""
if self.section != section:
if self.section > section:
raise dns.exception.FormError
self.section = section
@contextlib.contextmanager
def _track_size(self):
start = self.output.tell()
yield start
if self.output.tell() > self.max_size:
self._rollback(start)
raise dns.exception.TooBig
@contextlib.contextmanager
def _temporarily_seek_to(self, where):
current = self.output.tell()
try:
self.output.seek(where)
yield
finally:
self.output.seek(current)
def add_question(self, qname, rdtype, rdclass=dns.rdataclass.IN):
"""Add a question to the message."""
self._set_section(QUESTION)
with self._track_size():
qname.to_wire(self.output, self.compress, self.origin)
self.output.write(struct.pack("!HH", rdtype, rdclass))
self.counts[QUESTION] += 1
def add_rrset(self, section, rrset, **kw):
"""Add the rrset to the specified section.
Any keyword arguments are passed on to the rdataset's to_wire()
routine.
"""
self._set_section(section)
with self._track_size():
n = rrset.to_wire(self.output, self.compress, self.origin, **kw)
self.counts[section] += n
def add_rdataset(self, section, name, rdataset, **kw):
"""Add the rdataset to the specified section, using the specified
name as the owner name.
Any keyword arguments are passed on to the rdataset's to_wire()
routine.
"""
self._set_section(section)
with self._track_size():
n = rdataset.to_wire(name, self.output, self.compress, self.origin, **kw)
self.counts[section] += n
def add_opt(self, opt, pad=0, opt_size=0, tsig_size=0):
"""Add *opt* to the additional section, applying padding if desired. The
padding will take the specified precomputed OPT size and TSIG size into
account.
Note that we don't have reliable way of knowing how big a GSS-TSIG digest
might be, so we we might not get an even multiple of the pad in that case."""
if pad:
ttl = opt.ttl
assert opt_size >= 11
opt_rdata = opt[0]
size_without_padding = self.output.tell() + opt_size + tsig_size
remainder = size_without_padding % pad
if remainder:
pad = b"\x00" * (pad - remainder)
else:
pad = b""
options = list(opt_rdata.options)
options.append(dns.edns.GenericOption(dns.edns.OptionType.PADDING, pad))
opt = dns.message.Message._make_opt( # pyright: ignore
ttl, opt_rdata.rdclass, options
)
self.was_padded = True
self.add_rrset(ADDITIONAL, opt)
def add_edns(self, edns, ednsflags, payload, options=None):
"""Add an EDNS OPT record to the message."""
# make sure the EDNS version in ednsflags agrees with edns
ednsflags &= 0xFF00FFFF
ednsflags |= edns << 16
opt = dns.message.Message._make_opt( # pyright: ignore
ednsflags, payload, options
)
self.add_opt(opt)
def add_tsig(
self,
keyname,
secret,
fudge,
id,
tsig_error,
other_data,
request_mac,
algorithm=dns.tsig.default_algorithm,
):
"""Add a TSIG signature to the message."""
s = self.output.getvalue()
if isinstance(secret, dns.tsig.Key):
key = secret
else:
key = dns.tsig.Key(keyname, secret, algorithm)
tsig = dns.message.Message._make_tsig( # pyright: ignore
keyname, algorithm, 0, fudge, b"", id, tsig_error, other_data
)
(tsig, _) = dns.tsig.sign(s, key, tsig[0], int(time.time()), request_mac)
self._write_tsig(tsig, keyname)
def add_multi_tsig(
self,
ctx,
keyname,
secret,
fudge,
id,
tsig_error,
other_data,
request_mac,
algorithm=dns.tsig.default_algorithm,
):
"""Add a TSIG signature to the message. Unlike add_tsig(), this can be
used for a series of consecutive DNS envelopes, e.g. for a zone
transfer over TCP [RFC2845, 4.4].
For the first message in the sequence, give ctx=None. For each
subsequent message, give the ctx that was returned from the
add_multi_tsig() call for the previous message."""
s = self.output.getvalue()
if isinstance(secret, dns.tsig.Key):
key = secret
else:
key = dns.tsig.Key(keyname, secret, algorithm)
tsig = dns.message.Message._make_tsig( # pyright: ignore
keyname, algorithm, 0, fudge, b"", id, tsig_error, other_data
)
(tsig, ctx) = dns.tsig.sign(
s, key, tsig[0], int(time.time()), request_mac, ctx, True
)
self._write_tsig(tsig, keyname)
return ctx
def _write_tsig(self, tsig, keyname):
if self.was_padded:
compress = None
else:
compress = self.compress
self._set_section(ADDITIONAL)
with self._track_size():
keyname.to_wire(self.output, compress, self.origin)
self.output.write(
struct.pack("!HHI", dns.rdatatype.TSIG, dns.rdataclass.ANY, 0)
)
with prefixed_length(self.output, 2):
tsig.to_wire(self.output)
self.counts[ADDITIONAL] += 1
with self._temporarily_seek_to(10):
self.output.write(struct.pack("!H", self.counts[ADDITIONAL]))
def write_header(self):
"""Write the DNS message header.
Writing the DNS message header is done after all sections
have been rendered, but before the optional TSIG signature
is added.
"""
with self._temporarily_seek_to(0):
self.output.write(
struct.pack(
"!HHHHHH",
self.id,
self.flags,
self.counts[0],
self.counts[1],
self.counts[2],
self.counts[3],
)
)
def get_wire(self):
"""Return the wire format message."""
return self.output.getvalue()
def reserve(self, size: int) -> None:
"""Reserve *size* bytes."""
if size < 0:
raise ValueError("reserved amount must be non-negative")
if size > self.max_size:
raise ValueError("cannot reserve more than the maximum size")
self.reserved += size
self.max_size -= size
def release_reserved(self) -> None:
"""Release the reserved bytes."""
self.max_size += self.reserved
self.reserved = 0

2068
venv/Lib/site-packages/dns/resolver.py Normal file
View File

File diff suppressed because it is too large Load Diff

106
venv/Lib/site-packages/dns/reversename.py Normal file
View File

@@ -0,0 +1,106 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2006-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""DNS Reverse Map Names."""
import binascii
import dns.exception
import dns.ipv4
import dns.ipv6
import dns.name
ipv4_reverse_domain = dns.name.from_text("in-addr.arpa.")
ipv6_reverse_domain = dns.name.from_text("ip6.arpa.")
def from_address(
text: str,
v4_origin: dns.name.Name = ipv4_reverse_domain,
v6_origin: dns.name.Name = ipv6_reverse_domain,
) -> dns.name.Name:
"""Convert an IPv4 or IPv6 address in textual form into a Name object whose
value is the reverse-map domain name of the address.
*text*, a ``str``, is an IPv4 or IPv6 address in textual form
(e.g. '127.0.0.1', '::1')
*v4_origin*, a ``dns.name.Name`` to append to the labels corresponding to
the address if the address is an IPv4 address, instead of the default
(in-addr.arpa.)
*v6_origin*, a ``dns.name.Name`` to append to the labels corresponding to
the address if the address is an IPv6 address, instead of the default
(ip6.arpa.)
Raises ``dns.exception.SyntaxError`` if the address is badly formed.
Returns a ``dns.name.Name``.
"""
try:
v6 = dns.ipv6.inet_aton(text)
if dns.ipv6.is_mapped(v6):
parts = [str(byte) for byte in v6[12:]]
origin = v4_origin
else:
parts = [x for x in str(binascii.hexlify(v6).decode())]
origin = v6_origin
except Exception:
parts = [str(byte) for byte in dns.ipv4.inet_aton(text)]
origin = v4_origin
return dns.name.from_text(".".join(reversed(parts)), origin=origin)
def to_address(
name: dns.name.Name,
v4_origin: dns.name.Name = ipv4_reverse_domain,
v6_origin: dns.name.Name = ipv6_reverse_domain,
) -> str:
"""Convert a reverse map domain name into textual address form.
*name*, a ``dns.name.Name``, an IPv4 or IPv6 address in reverse-map name
form.
*v4_origin*, a ``dns.name.Name`` representing the top-level domain for
IPv4 addresses, instead of the default (in-addr.arpa.)
*v6_origin*, a ``dns.name.Name`` representing the top-level domain for
IPv4 addresses, instead of the default (ip6.arpa.)
Raises ``dns.exception.SyntaxError`` if the name does not have a
reverse-map form.
Returns a ``str``.
"""
if name.is_subdomain(v4_origin):
name = name.relativize(v4_origin)
text = b".".join(reversed(name.labels))
# run through inet_ntoa() to check syntax and make pretty.
return dns.ipv4.inet_ntoa(dns.ipv4.inet_aton(text))
elif name.is_subdomain(v6_origin):
name = name.relativize(v6_origin)
labels = list(reversed(name.labels))
parts = []
for i in range(0, len(labels), 4):
parts.append(b"".join(labels[i : i + 4]))
text = b":".join(parts)
# run through inet_ntoa() to check syntax and make pretty.
return dns.ipv6.inet_ntoa(dns.ipv6.inet_aton(text))
else:
raise dns.exception.SyntaxError("unknown reverse-map address family")

287
venv/Lib/site-packages/dns/rrset.py Normal file
View File

@@ -0,0 +1,287 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2003-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""DNS RRsets (an RRset is a named rdataset)"""
from typing import Any, Collection, Dict, cast
import dns.name
import dns.rdata
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.renderer
class RRset(dns.rdataset.Rdataset):
"""A DNS RRset (named rdataset).
RRset inherits from Rdataset, and RRsets can be treated as
Rdatasets in most cases. There are, however, a few notable
exceptions. RRsets have different to_wire() and to_text() method
arguments, reflecting the fact that RRsets always have an owner
name.
"""
__slots__ = ["name", "deleting"]
def __init__(
self,
name: dns.name.Name,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
deleting: dns.rdataclass.RdataClass | None = None,
):
"""Create a new RRset."""
super().__init__(rdclass, rdtype, covers)
self.name = name
self.deleting = deleting
def _clone(self):
obj = cast(RRset, super()._clone())
obj.name = self.name
obj.deleting = self.deleting
return obj
def __repr__(self):
if self.covers == 0:
ctext = ""
else:
ctext = "(" + dns.rdatatype.to_text(self.covers) + ")"
if self.deleting is not None:
dtext = " delete=" + dns.rdataclass.to_text(self.deleting)
else:
dtext = ""
return (
"<DNS "
+ str(self.name)
+ " "
+ dns.rdataclass.to_text(self.rdclass)
+ " "
+ dns.rdatatype.to_text(self.rdtype)
+ ctext
+ dtext
+ " RRset: "
+ self._rdata_repr()
+ ">"
)
def __str__(self):
return self.to_text()
def __eq__(self, other):
if isinstance(other, RRset):
if self.name != other.name:
return False
elif not isinstance(other, dns.rdataset.Rdataset):
return False
return super().__eq__(other)
def match(self, *args: Any, **kwargs: Any) -> bool: # type: ignore[override]
"""Does this rrset match the specified attributes?
Behaves as :py:func:`full_match()` if the first argument is a
``dns.name.Name``, and as :py:func:`dns.rdataset.Rdataset.match()`
otherwise.
(This behavior fixes a design mistake where the signature of this
method became incompatible with that of its superclass. The fix
makes RRsets matchable as Rdatasets while preserving backwards
compatibility.)
"""
if isinstance(args[0], dns.name.Name):
return self.full_match(*args, **kwargs) # type: ignore[arg-type]
else:
return super().match(*args, **kwargs) # type: ignore[arg-type]
def full_match(
self,
name: dns.name.Name,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType,
deleting: dns.rdataclass.RdataClass | None = None,
) -> bool:
"""Returns ``True`` if this rrset matches the specified name, class,
type, covers, and deletion state.
"""
if not super().match(rdclass, rdtype, covers):
return False
if self.name != name or self.deleting != deleting:
return False
return True
# pylint: disable=arguments-differ
def to_text( # type: ignore[override]
self,
origin: dns.name.Name | None = None,
relativize: bool = True,
**kw: Dict[str, Any],
) -> str:
"""Convert the RRset into DNS zone file format.
See ``dns.name.Name.choose_relativity`` for more information
on how *origin* and *relativize* determine the way names
are emitted.
Any additional keyword arguments are passed on to the rdata
``to_text()`` method.
*origin*, a ``dns.name.Name`` or ``None``, the origin for relative
names.
*relativize*, a ``bool``. If ``True``, names will be relativized
to *origin*.
"""
return super().to_text(
self.name, origin, relativize, self.deleting, **kw # type: ignore
)
def to_wire( # type: ignore[override]
self,
file: Any,
compress: dns.name.CompressType | None = None, # type: ignore
origin: dns.name.Name | None = None,
**kw: Dict[str, Any],
) -> int:
"""Convert the RRset to wire format.
All keyword arguments are passed to ``dns.rdataset.to_wire()``; see
that function for details.
Returns an ``int``, the number of records emitted.
"""
return super().to_wire(
self.name, file, compress, origin, self.deleting, **kw # type:ignore
)
# pylint: enable=arguments-differ
def to_rdataset(self) -> dns.rdataset.Rdataset:
"""Convert an RRset into an Rdataset.
Returns a ``dns.rdataset.Rdataset``.
"""
return dns.rdataset.from_rdata_list(self.ttl, list(self))
def from_text_list(
name: dns.name.Name | str,
ttl: int,
rdclass: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
text_rdatas: Collection[str],
idna_codec: dns.name.IDNACodec | None = None,
origin: dns.name.Name | None = None,
relativize: bool = True,
relativize_to: dns.name.Name | None = None,
) -> RRset:
"""Create an RRset with the specified name, TTL, class, and type, and with
the specified list of rdatas in text format.
*idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA
encoder/decoder to use; if ``None``, the default IDNA 2003
encoder/decoder is used.
*origin*, a ``dns.name.Name`` (or ``None``), the
origin to use for relative names.
*relativize*, a ``bool``. If true, name will be relativized.
*relativize_to*, a ``dns.name.Name`` (or ``None``), the origin to use
when relativizing names. If not set, the *origin* value will be used.
Returns a ``dns.rrset.RRset`` object.
"""
if isinstance(name, str):
name = dns.name.from_text(name, None, idna_codec=idna_codec)
rdclass = dns.rdataclass.RdataClass.make(rdclass)
rdtype = dns.rdatatype.RdataType.make(rdtype)
r = RRset(name, rdclass, rdtype)
r.update_ttl(ttl)
for t in text_rdatas:
rd = dns.rdata.from_text(
r.rdclass, r.rdtype, t, origin, relativize, relativize_to, idna_codec
)
r.add(rd)
return r
def from_text(
name: dns.name.Name | str,
ttl: int,
rdclass: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
*text_rdatas: Any,
) -> RRset:
"""Create an RRset with the specified name, TTL, class, and type and with
the specified rdatas in text format.
Returns a ``dns.rrset.RRset`` object.
"""
return from_text_list(
name, ttl, rdclass, rdtype, cast(Collection[str], text_rdatas)
)
def from_rdata_list(
name: dns.name.Name | str,
ttl: int,
rdatas: Collection[dns.rdata.Rdata],
idna_codec: dns.name.IDNACodec | None = None,
) -> RRset:
"""Create an RRset with the specified name and TTL, and with
the specified list of rdata objects.
*idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA
encoder/decoder to use; if ``None``, the default IDNA 2003
encoder/decoder is used.
Returns a ``dns.rrset.RRset`` object.
"""
if isinstance(name, str):
name = dns.name.from_text(name, None, idna_codec=idna_codec)
if len(rdatas) == 0:
raise ValueError("rdata list must not be empty")
r = None
for rd in rdatas:
if r is None:
r = RRset(name, rd.rdclass, rd.rdtype)
r.update_ttl(ttl)
r.add(rd)
assert r is not None
return r
def from_rdata(name: dns.name.Name | str, ttl: int, *rdatas: Any) -> RRset:
"""Create an RRset with the specified name and TTL, and with
the specified rdata objects.
Returns a ``dns.rrset.RRset`` object.
"""
return from_rdata_list(name, ttl, cast(Collection[dns.rdata.Rdata], rdatas))

118
venv/Lib/site-packages/dns/serial.py Normal file
View File

@@ -0,0 +1,118 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
"""Serial Number Arthimetic from RFC 1982"""
class Serial:
def __init__(self, value: int, bits: int = 32):
self.value = value % 2**bits
self.bits = bits
def __repr__(self):
return f"dns.serial.Serial({self.value}, {self.bits})"
def __eq__(self, other):
if isinstance(other, int):
other = Serial(other, self.bits)
elif not isinstance(other, Serial) or other.bits != self.bits:
return NotImplemented
return self.value == other.value
def __ne__(self, other):
if isinstance(other, int):
other = Serial(other, self.bits)
elif not isinstance(other, Serial) or other.bits != self.bits:
return NotImplemented
return self.value != other.value
def __lt__(self, other):
if isinstance(other, int):
other = Serial(other, self.bits)
elif not isinstance(other, Serial) or other.bits != self.bits:
return NotImplemented
if self.value < other.value and other.value - self.value < 2 ** (self.bits - 1):
return True
elif self.value > other.value and self.value - other.value > 2 ** (
self.bits - 1
):
return True
else:
return False
def __le__(self, other):
return self == other or self < other
def __gt__(self, other):
if isinstance(other, int):
other = Serial(other, self.bits)
elif not isinstance(other, Serial) or other.bits != self.bits:
return NotImplemented
if self.value < other.value and other.value - self.value > 2 ** (self.bits - 1):
return True
elif self.value > other.value and self.value - other.value < 2 ** (
self.bits - 1
):
return True
else:
return False
def __ge__(self, other):
return self == other or self > other
def __add__(self, other):
v = self.value
if isinstance(other, Serial):
delta = other.value
elif isinstance(other, int):
delta = other
else:
raise ValueError
if abs(delta) > (2 ** (self.bits - 1) - 1):
raise ValueError
v += delta
v = v % 2**self.bits
return Serial(v, self.bits)
def __iadd__(self, other):
v = self.value
if isinstance(other, Serial):
delta = other.value
elif isinstance(other, int):
delta = other
else:
raise ValueError
if abs(delta) > (2 ** (self.bits - 1) - 1):
raise ValueError
v += delta
v = v % 2**self.bits
self.value = v
return self
def __sub__(self, other):
v = self.value
if isinstance(other, Serial):
delta = other.value
elif isinstance(other, int):
delta = other
else:
raise ValueError
if abs(delta) > (2 ** (self.bits - 1) - 1):
raise ValueError
v -= delta
v = v % 2**self.bits
return Serial(v, self.bits)
def __isub__(self, other):
v = self.value
if isinstance(other, Serial):
delta = other.value
elif isinstance(other, int):
delta = other
else:
raise ValueError
if abs(delta) > (2 ** (self.bits - 1) - 1):
raise ValueError
v -= delta
v = v % 2**self.bits
self.value = v
return self